在之前的一篇博客中,我们安装了RabbitMQ并且对其也有的初步的认识,今天就来写个入门小例子来加深概念理解并了解代码怎么实现。
本篇博客围绕下面几个方面展开:
- 代码前的理论热身
- 代码实例:Hello RabbitMQ
- 运行代码并调试问题
Now, Let's begin !
一、代码前的理论热身
我们来看张图:
Publisher(生产者)生成消息,然后publish(发布)消息到exchange(路由器,也有资料翻译成交换机),然后根据路由规则将消息传递到Queue(队列),最终交由Consumer(消费者)进行消费处理。
这里的生产者和消费者都是我们的应用,因此我们的代码中要实现这两个部分。
中间的节点就是RabbitMQ 提供的内容,需要再生产者和消费者里面调用其接口来定义和使用这些节点。
二、代码实例:Hello RabbitMQ
- 首先来实现生产者,这里我没有用Publisher做类名,而是用的Provider,没有特别的用意,就是在起名字的时候不小心写成了这样,不需要在意这个细节,O(∩_∩)O。
package com.sam.hello_rabbitmq;import java.io.IOException;import java.util.concurrent.TimeoutException;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;public class Provider { //定义队列名 static String QUEUE_NAME = "helloRabbit"; public static void main(String[] args) { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = null; Channel channel = null; try { //1.创建连接和通道 connection = factory.newConnection(); channel = connection.createChannel(); //2.为通道声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //3.发布消息 String msg = " hello rabbitmq, welcome to sam's blog."; channel.basicPublish("", QUEUE_NAME, null, msg.getBytes()); System.out.println("provider send a msg: " + msg); } catch (IOException e) { e.printStackTrace(); } catch (TimeoutException e) { e.printStackTrace(); } finally { //4.关闭连接 if (channel != null) { try { channel.close(); } catch (IOException e) { e.printStackTrace(); } catch (TimeoutException e) { e.printStackTrace(); } } if (connection != null) { try { connection.close(); } catch (IOException e) { e.printStackTrace(); } } } }}
在第2步中,channel.queueDeclare 用来创建队列,有5个参数:String queue, 队列名; boolean durable, 该队列是否需要持久化; boolean exclusive,该队列是否为该通道独占的(其他通道是否可以消费该队列); boolean autoDelete,该队列不再使用的时候,是否让RabbitMQ服务器自动删除掉; Map<String, Object> arguments 其他参数。第3步中,channel.basicPublish 发布消息(用在生产者),有4个参数:String exchange, 路由器(有的资料翻译成交换机)的名字,即将消息发到哪个路由器; String routingKey, 路由键,即发布消息时,该消息的路由键是什么; BasicProperties props, 指定消息的基本属性; byte[] body 消息体,也就是消息的内容,是字节数组。 可能你会疑惑,为什么没有exchange呢?因为如果声明了队列,可以不声明路由器。
- 接着来实现消费者,消费者实现和生产者过程差不多,但是在这里并没有关闭连接和通道,是因为要消费者一直等待随时可能发来的消息。代码如下:
package com.sam.hello_rabbitmq;import java.io.IOException;import java.util.concurrent.TimeoutException;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.Consumer;import com.rabbitmq.client.DefaultConsumer;import com.rabbitmq.client.Envelope;public class HelloConsumer { public static void main(String[] args) { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = null; Channel channel = null; try { // 1.创建连接和通道 connection = factory.newConnection(); channel = connection.createChannel(); // 2.为通道声明队列 channel.queueDeclare(Provider.QUEUE_NAME, false, false, false, null); System.out.println(" **** keep alive ,waiting for messages, and then deal them"); // 3.通过回调生成消费者 Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, com.rabbitmq.client.AMQP.BasicProperties properties, byte[] body) throws IOException { //获取消息内容然后处理 String msg = new String(body, "UTF-8"); System.out.println("*********** HelloConsumer" + " get message :[" + msg +"]"); } }; //4.消费消息 channel.basicConsume(Provider.QUEUE_NAME, true, consumer); } catch (IOException e) { e.printStackTrace(); } catch (TimeoutException e) { e.printStackTrace(); } }}
在第4步中,channel.basicConsume 用来接收消息,用在消费者,有3个参数:String queue, 队列名字,即要从哪个队列中接收消息; boolean autoAck, 是否自动确认,默认true; Consumer callback 消费者,即谁接收消息。
三、运行代码并调试问题
代码写好了,接下来进行测试,
- 先来执行下Provider.java,发现报错了:
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".SLF4J: Defaulting to no-operation (NOP) logger implementationSLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.java.io.IOException at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:124) at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:120) at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:142) at com.rabbitmq.client.impl.ChannelN.queueDeclare(ChannelN.java:952) at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.queueDeclare(AutorecoveringChannel.java:333) at com.sam.hello_rabbitmq.Provider.main(Provider.java:36)Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method
(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'helloRabbit' in vhost '/': received 'false' but current is 'true', class-id=50, method-id=10) at com.rabbitmq.utility.ValueOrException.getValue(ValueOrException.java:66) at com.rabbitmq.utility.BlockingValueOrException.uninterruptibleGetValue(BlockingValueOrException.java:36) at com.rabbitmq.client.impl.AMQChannel$BlockingRpcContinuation.getReply(AMQChannel.java:443) at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:263) at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:136) ... 3 moreCaused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method (reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'helloRabbit' in vhost '/': received 'false' but current is 'true', class-id=50, method-id=10) at com.rabbitmq.client.impl.ChannelN.asyncShutdown(ChannelN.java:509) at com.rabbitmq.client.impl.ChannelN.processAsync(ChannelN.java:340) at com.rabbitmq.client.impl.AMQChannel.handleCompleteInboundCommand(AMQChannel.java:162) at com.rabbitmq.client.impl.AMQChannel.handleFrame(AMQChannel.java:109) at com.rabbitmq.client.impl.AMQConnection.readFrame(AMQConnection.java:643) at com.rabbitmq.client.impl.AMQConnection.access$300(AMQConnection.java:47) at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:581) at java.lang.Thread.run(Thread.java:745)Exception in thread "main" com.rabbitmq.client.AlreadyClosedException: channel is already closed due to channel error; protocol method: #method (reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'helloRabbit' in vhost '/': received 'false' but current is 'true', class-id=50, method-id=10) at com.rabbitmq.client.impl.AMQChannel.processShutdownSignal(AMQChannel.java:345) at com.rabbitmq.client.impl.ChannelN.startProcessShutdownSignal(ChannelN.java:286) at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:600) at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:534) at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:527) at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.close(AutorecoveringChannel.java:68) at com.sam.hello_rabbitmq.Provider.main(Provider.java:60) 关键堆栈信息是:inequivalent arg 'durable' for queue 'helloRabbit' in vhost '/': received 'false' but current is 'true',说是helloRabbit这个队列durable(是否需要持久化) 参数已经设定成了true 但是代码中指定的是false,冲突了,纳尼?访问RabbitMQ管理页面:http://localhost:15672/#/queues 发现已经存在一个队列helloRabbit,
点helloRabbit的链接,发现队列的durable属性确实是true。哦,原来我之前在做别的练习的时候,创建过一个叫这个名字的队列了,而且属性值刚好为true.
那么接下来删掉这个既存的队列
再去执行Provider.java,后台打印了内容,并且队列中有了一条ready的消息。
问题解决!
- 执行HelloConsumer.java,预想的结果是在启动后,控制台直接打印出log并且RabbitMQ管理页面没有ready的消息:
结果符合预期。
到此,全部工作完美结束。