RabbitMQ消息队列之二:RabbitMQ的入门使用

在使用RabbitMQ之前,需要了解RabbitMQ的工作原理。

RabbitMQ的工作原理

RabbitMQ是消息代理。从本质上说,它接受来自生产者的信息,并将它们传递给消费者。在两者之间,它可以根据你给它的路由,缓冲规则进行传递消息。

示例图

这里写图片描述

RabbitMQ的术语解释

生产者:生产消息,发送消息。类似工厂。
消费者:接受消息,使用消息。类似顾客。
队列:存储消息。类似仓库、中转站。队列可以存储很多的消息,因为它基本上是一个无限制的缓冲区,前提是你的机器有足够的存储空间。多个生产者可以将消息发送到同一个队列中,多个消费者也可以只从同一个队列接收数据。这就是队列的特性。

RabbitMQ 消息生产者的代码实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
import java.util.HashMap;
import java.util.Map;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
public class RabbitProducer {
private final static String QUEUE_NAME = "RabbitMQ_Hello"; //消息队列名
public static void main(String[] argv) throws Exception {
//创建连接连接到RabbitMQ
ConnectionFactory factory = new ConnectionFactory();
// 设置ip
factory.setHost("127.0.0.1");
/* //设置端口
factory.setPort(15672);
//设置用户名
factory.setUsername("guest");
//设置密码
factory.setPassword("guest");
//设置url(包括ip、端口、用户名、密码)
factory.setUri("amqp://guest:guest@localhost:15672");
*/
// 创建一个连接
Connection connection = factory.newConnection();
// 创建一个频道
Channel channel = connection.createChannel();
// 指定一个队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
Map<String,Object> map=new HashMap<String,Object>();
map.put("java", "hello");
map.put("RabbitMQ", "Hello");
//发送的消息
String message = JSON.toJSONString(map);
// 往队列中发出一条消息
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
// 关闭频道和连接
channel.close();
connection.close();
}
}

RabbitMQ消息消费者的代码实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
//消费者
public class RabbitConsumer {
private final static String QUEUE_NAME = "RabbitMQ_Hello"; //消息队列名
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
// 打开连接和创建频道,与发送端一样
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 consumer = new QueueingConsumer(channel);
// 指定消费队列
channel.basicConsume(QUEUE_NAME, true, consumer);
while (true) { //消费者程序运行开着 如果生产者新增了数据会自动获取
Thread.sleep(500);
// nextDelivery是一个阻塞方法(内部实现其实是阻塞队列的take方法)
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println("'[x] Received '" + message );
}
}
}

打开本地的RabbitMQ,之后运行生产者代码,会将消息存储到RabbitMQ中,可以通过RabbitMQ web界面看到消息。

这里写图片描述

这里写图片描述

这里写图片描述

这里写图片描述

运行消费者代码,会将该消息消费掉,可以在RabbitMQ web界面可以发现上面存储的消息已经没有了。
这里写图片描述

这里写图片描述

通过这个demo,可以对消息队列有了一个的基本的了解了。
版权声明:
作者:虚无境
博客园出处:http://www.cnblogs.com/xuwujing
CSDN出处:http://blog.csdn.net/qazwsxpcm    
个人博客出处:http://www.panchengming.com
原创不易,转载请标明出处,谢谢!

+
------ 本文结束 ------