RabbitMQ学习笔记二：Java实现RabbitMQ

远行的心

　　本地安装好RabbitMQ Server后，就可以在Java语言中使用RabbitMQ了。
　　RabbitMQ是一个消息代理，从“生产者”接收消息并传递消息至“消费者”，期间可根据规则路由、缓存、持久化消息。“生产者”也即message发送者以下简称P，相对应的“消费者”乃message接收者以下简称C，message通过queue由P到C，queue存在于RabbitMQ，可存储尽可能多的message，多个P可向同一queue发送message，多个C可从同一queue接收message。
　　几个关键概念:

Broker：简单来说就是消息队列服务器实体。
Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。
Queue：消息队列载体，每个消息都会被投入到一个或多个队列。
Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。
Routing Key：路由关键字，exchange根据这个关键字进行消息投递。
vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。
producer：消息生产者，就是投递消息的程序。
consumer：消息消费者，就是接受消息的程序。
channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。

由Exchange，Queue，RoutingKey三个才能决定一个从Exchange到Queue的唯一的线路。

消息队列的使用过程大概如下：

（1）客户端连接到消息队列服务器，打开一个channel。
（2）客户端声明一个exchange，并设置相关属性。
（3）客户端声明一个queue，并设置相关属性。
（4）客户端使用routing key，在exchange和queue之间建立好绑定关系。
（5）客户端投递消息到exchange。

　　现在，可以上代码了。首先，是在项目中加入需要的jar包，我使用的是maven项目，直接配置maven及可：



<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>3.0.4</version>
</dependency>
　　后面还会用到的jar包，配置如下



<dependency>
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
<version>2.6</version>
</dependency>
　　先写一个类，将产生产者和消费者统一为 EndPoint类型的队列。不管是生产者还是消费者， 连接队列的代码都是一样的，这样可以通用一些。



package cn.com.shopec.rabbitmq;
import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public abstract class EndPoint {
protected Channel channel;
protected Connection connection;
protected String endPointName;
public EndPoint(String endpointName) throws IOException
{
this.endPointName = endpointName;
// Create a connection factory
ConnectionFactory factory = new ConnectionFactory();
// 与RabbitMQ Server建立连接  
// 连接到的broker在本机localhost上
factory.setHost("localhost");
// getting a connection
connection = factory.newConnection();
// creating a channel
channel = connection.createChannel();
// declaring a queue for this channel. If queue does not exist,
// it will be created on the server.
// queueDeclare的参数：queue 队列名；durable true为持久化；exclusive 是否排外，true为队列只可以在本次的连接中被访问，
// autoDelete true为connection断开队列自动删除；arguments 用于拓展参数
channel.queueDeclare(endpointName, false, false, false, null);
}
/**
* 关闭channel和connection。并非必须，因为隐含是自动调用的。
* @throws IOException
*/
public void close() throws IOException
{
this.channel.close();
this.connection.close();
}
}
　　生产者类的任务是向队列里写一条消息



package cn.com.shopec.rabbitmq;
import java.io.IOException;
import java.io.Serializable;
import org.apache.commons.lang.SerializationUtils;
public class Producer extends EndPoint {
public Producer(String endPointName) throws IOException
{
super(endPointName);
}
public void sendMessage(Serializable object) throws IOException
{
channel.basicPublish("", endPointName, null, SerializationUtils.serialize(object));
}
}
　　消费者可以以线程方式运行，对于不同的事件有不同的回调函数，其中最主要的是处理新消息到来的事件。



package cn.com.shopec.rabbitmq;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.lang.SerializationUtils;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;
public class QueueConsumer extends EndPoint implements Runnable, Consumer {
public QueueConsumer(String endPointName) throws IOException
{
super(endPointName);
}
public void run()
{
try
{
// start consuming messages. Auto acknowledge messages.
channel.basicConsume(endPointName, true, this);
}
catch (IOException e)
{
e.printStackTrace();
}
}
/**
* Called when consumer is registered.
*/
public void handleConsumeOk(String consumerTag)
{
System.out.println("Consumer " + consumerTag + " registered");
}
/**
* Called when new message is available.
*/
public void handleDelivery(String consumerTag, Envelope env, BasicProperties props, byte[] body) throws IOException
{
Map map = (HashMap) SerializationUtils.deserialize(body);
System.out.println("Message Number " + map.get("message number") + " received.");
}
public void handleCancel(String consumerTag)
{
}
public void handleCancelOk(String consumerTag)
{
}
public void handleRecoverOk(String consumerTag)
{
}
public void handleShutdownSignal(String consumerTag, ShutdownSignalException arg1)
{
}
}
　　测试类中，先运行一个消费者线程，然后开始产生大量的消息，这些消息会被消费者取走。



package cn.com.shopec.rabbitmq;
import java.io.IOException;
import java.sql.SQLException;
import java.util.HashMap;
public class Main {
public Main() throws Exception
{
// 创建消费者，即消息接收者，并启动线程
QueueConsumer consumer = new QueueConsumer("queue");
Thread consumerThread = new Thread(consumer);
consumerThread.start();
// 创建生产者，即消息发送者
Producer producer = new Producer("queue");
// 循环发送消息
for (int i = 0; i < 20; i++)
{
HashMap message = new HashMap();
message.put("message number", i);
producer.sendMessage(message);
System.out.println("Message Number " + i + " sent.");
}
}
/**
* @param args
* @throws SQLException
* @throws IOException
*/
public static void main(String[] args) throws Exception
{
new Main();
}
}
　　运行结果：
　　Consumer amq.ctag-8TFduKUwrE1I8iT2L5DaZg registered
Message Number 0 sent.
Message Number 1 sent.
Message Number 2 sent.
Message Number 3 sent.
Message Number 4 sent.
Message Number 5 sent.
Message Number 6 sent.
Message Number 7 sent.
Message Number 8 sent.
Message Number 9 sent.
Message Number 10 sent.
Message Number 11 sent.
Message Number 12 sent.
Message Number 13 sent.
Message Number 14 sent.
Message Number 15 sent.
Message Number 16 sent.
Message Number 17 sent.
Message Number 18 sent.
Message Number 19 sent.
Message Number 0 received.
Message Number 1 received.
Message Number 2 received.
Message Number 3 received.
Message Number 4 received.
Message Number 5 received.
Message Number 6 received.
Message Number 7 received.
Message Number 8 received.
Message Number 9 received.
Message Number 10 received.
Message Number 11 received.
Message Number 12 received.
Message Number 13 received.
Message Number 14 received.
Message Number 15 received.
Message Number 16 received.
Message Number 17 received.
Message Number 18 received.
Message Number 19 received.