RabbitMQ-从基础到实战— Hello RabbitMQ
Posted sharpest
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RabbitMQ-从基础到实战— Hello RabbitMQ相关的知识,希望对你有一定的参考价值。
转自:https://yq.aliyun.com/articles/589923
1.简介
本篇博文介绍了在windows平台下安装RabbitMQ Server端,并用JAVA代码实现收发消息
2.安装RabbitMQ
Windows平台安装完成后如图
3.启用RabbitMQ Web控制台
RabbitMQ提供一个控制台,用于管理和监控RabbitMQ,默认是不启动的,需要运行以下命令进行启动
- 点击上图的Rabbit Command Prompt,打开rabbitMQ控制台
- 在官方介绍管理控制台的页面,可以看到,输入以下命令启动后台控制插件
rabbitmq-plugins enable rabbitmq_management
- 登录后台页面:http://localhost:15672/ 密码和用户名都是 guest ,界面如下
目前可以先不用理会此界面,后面使用到时会详细介绍,也可以到这里查看官方文档。
4.编写MessageSender
Spring对RabbitMQ已经进行了封装,正常使用中,会使用Spring集成,第一个项目中,我们先不考虑那么多
在IDE中新建一个Maven项目,并在pom.xml中贴入如下依赖,RabbitMQ的最新版本依赖可以在这里找到
<dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>4.1.0</version> </dependency>
等待Maven下载完成后,就可以在Maven Dependencies中看到RabbitMQ的JAR
在这里,我们发现,RabbitMQ的日志依赖了slf4j-api这个包,slf4j-api并不是一个日志实现,这样子是打不出日志的,所以,我们给pom加上一个日志实现,这里用了logback
<dependency> <groupId>ch.qos.logback</groupId> <artifactId>logback-classic</artifactId> <version>1.2.1</version> </dependency>
之后maven依赖如下,可以放心写代码了
新建一个MessageSender类,代码如下
1 import java.io.IOException; 2 import java.util.concurrent.TimeoutException; 3 4 import org.slf4j.Logger; 5 import org.slf4j.LoggerFactory; 6 7 import com.rabbitmq.client.Channel; 8 import com.rabbitmq.client.Connection; 9 import com.rabbitmq.client.ConnectionFactory; 10 11 public class MessageSender { 12 13 private Logger logger = LoggerFactory.getLogger(MessageSender.class); 14 15 //声明一个队列名字 16 private final static String QUEUE_NAME = "hello"; 17 18 public boolean sendMessage(String message){ 19 //new一个RabbitMQ的连接工厂 20 ConnectionFactory factory = new ConnectionFactory(); 21 //设置需要连接的RabbitMQ地址,这里指向本机 22 factory.setHost("127.0.0.1"); 23 Connection connection = null; 24 Channel channel = null; 25 try { 26 //尝试获取一个连接 27 connection = factory.newConnection(); 28 //尝试创建一个channel 29 channel = connection.createChannel(); 30 //这里的参数在后面详解 31 channel.queueDeclare(QUEUE_NAME, false, false, false, null); 32 //注意这里调用了getBytes(),发送的其实是byte数组,接收方收到消息后,需要重新组装成String 33 channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); 34 logger.info("Sent ‘" + message + "‘"); 35 //关闭channel和连接 36 channel.close(); 37 connection.close(); 38 } catch (IOException | TimeoutException e) { 39 //失败后记录日志,返回false,代表发送失败 40 logger.error("send message faild!",e); 41 return false; 42 } 43 return true; 44 } 45 }
然后在App类的main方法中调用sendMessage
1 public class App { 2 public static void main( String[] args ){ 3 MessageSender sender = new MessageSender(); 4 sender.sendMessage("hello RabbitMQ!"); 5 } 6 }
打印日志如下
打开RabbitMQ的控制台,可以看到消息已经进到了RabbitMQ中
点进去,用控制台自带的getMessage功能,可以看到消息已经成功由RabbitMQ管理了
至此,MessageSender已经写好了,在该类的31和33行,我们分别调用了队列声明和消息发送
channel.queueDeclare(QUEUE_NAME, false, false, false, null); channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
queueDeclare,有很多参数,我们可以看一下他的源码,注释上有详细的解释,我简单翻译了一下
1 /** 2 * Declare a queue 声明一个队列 3 * @see com.rabbitmq.client.AMQP.Queue.Declare 4 * @see com.rabbitmq.client.AMQP.Queue.DeclareOk 5 * @param queue the name of the queue队列的名字 6 * @param durable true if we are declaring a durable queue (the queue will survive a server restart)是否持久化,为true则在rabbitMQ重启后生存 7 * @param exclusive true if we are declaring an exclusive queue (restricted to this connection)是否是排他性队列(别人看不到),只对当前连接有效,当前连接断开后,队列删除(设置了持久化也删除) 8 * @param autoDelete true if we are declaring an autodelete queue (server will delete it when no longer in use)自动删除,在最后一个连接断开后删除队列 9 * @param arguments other properties (construction arguments) for the queue 其他参数 10 * @return a declaration-confirm method to indicate the queue was successfully declared 11 * @throws java.io.IOException if an error is encountered 12 */ 13 Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, 14 Map<String, Object> arguments) throws IOException;
前面4个都非常好理解,最后一个“其他参数”,到底是什么其他参数,这个东西真的很难找,用到再解释吧,官方文档如下
- TTL Time To Live 存活时间
- Dead Lettering 遗言,当消息死亡时,做些什么
- Length Limit 长度限制
- Priority Queues 优先级
basicPublish的翻译如下
1 /** 2 * Publish a message.发送一条消息 3 * 4 * Publishing to a non-existent exchange will result in a channel-level 5 * protocol exception, which closes the channel. 6 * 7 * Invocations of <code>Channel#basicPublish</code> will eventually block if a 8 * <a href="http://www.rabbitmq.com/alarms.html">resource-driven alarm</a> is in effect. 9 * 10 * @see com.rabbitmq.client.AMQP.Basic.Publish 11 * @see <a href="http://www.rabbitmq.com/alarms.html">Resource-driven alarms</a> 12 * @param exchange the exchange to publish the message to 交换模式,会在后面讲,官方文档在这里 13 * @param routingKey the routing key 控制消息发送到哪个队列 14 * @param props other properties for the message - routing headers etc 其他参数 15 * @param body the message body 消息,byte数组 16 * @throws java.io.IOException if an error is encountered 17 */ 18 void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;
这里又有个其他参数,它的类型是这样的,设置消息的一些详细属性
5.编写MessageConsumer
为了和Sender区分开,新建一个Maven项目MessageConsumer
1 package com.liyang.ticktock.rabbitmq; 2 3 import java.io.IOException; 4 import java.util.concurrent.TimeoutException; 5 6 import org.slf4j.Logger; 7 import org.slf4j.LoggerFactory; 8 9 import com.rabbitmq.client.AMQP; 10 import com.rabbitmq.client.Channel; 11 import com.rabbitmq.client.Connection; 12 import com.rabbitmq.client.ConnectionFactory; 13 import com.rabbitmq.client.Consumer; 14 import com.rabbitmq.client.DefaultConsumer; 15 import com.rabbitmq.client.Envelope; 16 17 public class MessageConsumer { 18 19 private Logger logger = LoggerFactory.getLogger(MessageConsumer.class); 20 21 public boolean consume(String queueName){ 22 //连接RabbitMQ 23 ConnectionFactory factory = new ConnectionFactory(); 24 factory.setHost("127.0.0.1"); 25 Connection connection = null; 26 Channel channel = null; 27 try { 28 connection = factory.newConnection(); 29 channel = connection.createChannel(); 30 //这里声明queue是为了取消息的时候,queue肯定会存在 31 //注意,queueDeclare是幂等的,也就是说,消费者和生产者,不论谁先声明,都只会有一个queue 32 channel.queueDeclare(queueName, false, false, false, null); 33 34 //这里重写了DefaultConsumer的handleDelivery方法,因为发送的时候对消息进行了getByte(),在这里要重新组装成String 35 Consumer consumer = new DefaultConsumer(channel){ 36 @Override 37 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) 38 throws IOException { 39 String message = new String(body, "UTF-8"); 40 logger.info("Received ‘" + message + "‘"); 41 } 42 }; 43 //上面是声明消费者,这里用声明的消费者消费掉队列中的消息 44 channel.basicConsume(queueName, true, consumer); 45 46 //这里不能关闭连接,调用了消费方法后,消费者会一直连接着rabbitMQ等待消费 47 48 } catch (IOException | TimeoutException e) { 49 //失败后记录日志,返回false,代表消费失败 50 logger.error("send message faild!",e); 51 return false; 52 } 53 54 55 return true; 56 } 57 }
然后在App的main方法中调用Cunsumer进行消费
1 public class App 2 { 3 //这个队列名字要和生产者中的名字一样,否则找不到队列 4 private final static String QUEUE_NAME = "hello"; 5 6 public static void main( String[] args ) 7 { 8 MessageConsumer consumer = new MessageConsumer(); 9 consumer.consume(QUEUE_NAME); 10 } 11 }
结果如下,消费者一直在等待消息,每次有消息进来,就会立刻消费掉
6.多个消费者同时消费一个队列
改造一下Consumer
在App中new多个消费者
改造Sender,使它不停的往RabbitMQ中发送消息
启动Sender
启动Consumer,发现消息很平均的发给四个客户端,一人一个,谁也不插队
如果我们把速度加快呢?把Sender的休息时间去掉,发现消费开始变得没有规律了,其实呢,它还是有规律的,这个是RabbitMQ的特性,称作“Round-robin dispatching”,消息会平均的发送给每一个消费者,可以看第一第二行,消息分别是56981和56985,相应的82、82、84都被分给了其他线程,只是在当前线程的时间片内,可以处理这么多任务,所以就一次打印出来了
以上是关于RabbitMQ-从基础到实战— Hello RabbitMQ的主要内容,如果未能解决你的问题,请参考以下文章