博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
RabbitMQ入门:Hello RabbitMQ 代码实例
阅读量:5298 次
发布时间:2019-06-14

本文共 8310 字,大约阅读时间需要 27 分钟。

在之前的一篇博客中,我们安装了RabbitMQ并且对其也有的初步的认识,今天就来写个入门小例子来加深概念理解并了解代码怎么实现。

本篇博客围绕下面几个方面展开:

  1. 代码前的理论热身
  2. 代码实例:Hello RabbitMQ
  3. 运行代码并调试问题

Now, Let's begin !

一、代码前的理论热身

我们来看张图:

Publisher(生产者)生成消息,然后publish(发布)消息到exchange(路由器,也有资料翻译成交换机),然后根据路由规则将消息传递到Queue(队列),最终交由Consumer(消费者)进行消费处理。

这里的生产者和消费者都是我们的应用,因此我们的代码中要实现这两个部分。

中间的节点就是RabbitMQ 提供的内容,需要再生产者和消费者里面调用其接口来定义和使用这些节点。

 

二、代码实例:Hello RabbitMQ

  1. 首先来实现生产者,这里我没有用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呢?因为如果声明了队列,可以不声明路由器。

  2. 接着来实现消费者,消费者实现和生产者过程差不多,但是在这里并没有关闭连接和通道,是因为要消费者一直等待随时可能发来的消息。代码如下:
    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 消费者,即谁接收消息。

 

三、运行代码并调试问题

代码写好了,接下来进行测试,

  1. 先来执行下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的消息。

    问题解决!

  2. 执行HelloConsumer.java,预想的结果是在启动后,控制台直接打印出log并且RabbitMQ管理页面没有ready的消息:

    结果符合预期。

到此,全部工作完美结束。

转载于:https://www.cnblogs.com/sam-uncle/p/9188426.html

你可能感兴趣的文章
Spring中事务传播行为
查看>>
java基础(一):我对java的三个环境变量的简单理解和配置
查看>>
arcgis api 4.x for js 结合 Echarts4 实现散点图效果(附源码下载)
查看>>
YTU 2734: 国家排序
查看>>
YTU 2625: B 构造函数和析构函数
查看>>
Notepad++ 16进制编辑功能
查看>>
Caffe: Cannot create Cublas handle. Cublas won't be available
查看>>
Linux 下 LXD 容器搭建 Hadoop 集群
查看>>
mysql describe
查看>>
Hello博客园
查看>>
apache自带压力测试工具ab的使用及解析
查看>>
Android基础入门教程——8.1.2 Android中的13种Drawable
查看>>
C语言作业3
查看>>
.Net Core中的通用主机(二)——托管服务
查看>>
C#使用Xamarin开发可移植移动应用(2.Xamarin.Forms布局,本篇很长,注意)附源码
查看>>
koogra--Excel文件读取利器
查看>>
ASP.NET 使用ajaxupload.js插件出现上传较大文件失败的解决方法
查看>>
jenkins搭建
查看>>
C#中使用Split分隔字符串的技巧
查看>>
(springboot)freemarker(二)
查看>>