1、相关概念
RabbitMQ是一个消息代理,事实上,它接收生产者产生的消息,然后将消息传递给消费者。在这个过程中,它可以路由,可以缓冲,或者更具你设定的规则来将消息持久化。RabbitMQ和消息传输过程中一般会用一些术语:
意思无非是指发送消息的那一端,如果一个程序发送消息,那么它就将被称为生产者,这里用大写的P来表示。
相当于邮箱的名字,它活动在RabbitMQ服务器里边。虽然消息流会通过RabbitMQ和你的应用程序,但是只会被存储在队列中。队列是不受任何限制的,它可以尽可能多的去存储你需要存储的消息(本质上来说它是个无限缓冲)。可以多个生产者向同一个消息队列发送消息,也可以多个消费者同时从一个消息队列中来接收消息。消息队列可以如下图模型。
即接收端,消费者主要是等待接收消息的程序,用下图表示:
注意:在大多数应用场景中,生产者、消费者以及RabbitMQ服务是不会同时运行在一台机器上的。
下边将会实现两个Java程序,一个只发送生产者一条消息的生产者,一个接收消息、并打印消息的消费者。
在下边的对话中,”P”是我们的生产者,”C”是我们的消费者,中间的是矩形是队列(BabbitMQ维护的消息缓冲)
2、Java客户端的包
RabbitMQ遵守AMQP协议,AMQP协议是一个开放、通用的消息协议。关于AMQP协议的客户端,有多种语言的实现版本。本文使用的是RabbitMQ提供的Java客户端。
可在http://www.rabbitmq.com/java-client.html这里下载,然后将相对应的Jar包拷贝你的工作目录下。RabbitMQ的Java客户端在Maven库中也有,groupId是com.rabbitmq,artifactId是amqp-client。
3、发送端
我们将消息发送器起名为Send,消息接收器起名为Recv。发送器将会连接RabbitMQ,发送一条消息,然后退出。
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
public class Send {
private final static String QUEUE_NAME = "hello";
public static void main(String[] argv) throws java.io.IOException {
/*连接可以抽象为socket连接,为我们维护协议版本信息和协议证书等。这里我们连接
上了本机的消息服务器实体(localhost)。如果我们想连接其它主机上的RabbitMQ服务,只需要修改一下主机名或是IP就可以了*/
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
/*接下创建channel(信道),这是绝大多数API都能用到的。为了发送消息,你必须要声明一个消息消息队列,然后向该队列里推送消息*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
String message = "Hello World!";
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent ‘" + message + "‘");
/*声明一个幂等的队列(只有在该队列不存在时,才会被创建)。消息的上下文是一个
字节数组,你可以指定它的编码。*/
channel.close();
connection.close();
}
}
注:若RabbitMQ服务磁盘空间不足的话,运行会出错,参数设置为:disk_free_limit,更多的参数配置可在这里找到http://www.rabbitmq.com/configure.html#config-items
4、接收端
RabbitMQ会往接收器上推消息,与只发送一条消息的发送端不同,这里我们将监听消息并将消息打印出来。
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
public class Recv {
private final static String QUEUE_NAME = "hello";
public static void main(String[] argv) throws java.io.IOException, java.lang.InterruptedException {
/*这里怎么打开连接和信道,以及声明用于接收消息的队列,这些步骤与发送端基本上是一样的*/
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
/*确保这里的队列是存在的*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
/*这里用到了额外的类QueueingConsumer来缓存服务器将要推过来的消息。我们通知服务器向接收端推送消息,
然后服务器将会向客户端异步推送消息,这里提供了一个可以回调的对象来缓存消息,直到我们做好准备来使用
它,这个类就是QueueingConsumer*/
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(QUEUE_NAME, true, consumer);
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" [x] Received ‘" + message + "‘");
}
}
}
消息接收完后QueueingConsumer.nextDelivery()将会发生阻塞,暂停运行,直到有其他的消息推过来。
如果你需要检查和验证队列的话,需要使用rabbitmqctl list_queues。
参考链接:http://www.rabbitmq.com/tutorials/tutorial-one-java.html
RabbitMQ系列 第二篇:快速入门HelloWorld