RabbitMQ入门:Hello RabbitMQ 代码实例

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

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

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

Now, Let's begin !

一、代码前的理论热身

我们来看张图:文章来源地址https://www.yii666.com/article/754387.html

RabbitMQ入门:Hello RabbitMQ 代码实例

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 implementation
    SLF4J: 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<channel.close>(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 more
    Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(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<channel.close>(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,

    RabbitMQ入门:Hello RabbitMQ 代码实例

    点helloRabbit的链接,发现队列的durable属性确实是true。哦,原来我之前在做别的练习的时候,创建过一个叫这个名字的队列了,而且属性值刚好为true.

    RabbitMQ入门:Hello RabbitMQ 代码实例

    那么接下来删掉这个既存的队列

    RabbitMQ入门:Hello RabbitMQ 代码实例网址:yii666.com

    再去执行Provider.java,后台打印了内容,并且队列中有了一条ready的消息。RabbitMQ入门:Hello RabbitMQ 代码实例RabbitMQ入门:Hello RabbitMQ 代码实例

    问题解决!网址:yii666.com<

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

    RabbitMQ入门:Hello RabbitMQ 代码实例文章地址https://www.yii666.com/article/754387.html文章来源地址:https://www.yii666.com/article/754387.html

    结果符合预期。

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

版权声明:本文内容来源于网络,版权归原作者所有,此博客不拥有其著作权,亦不承担相应法律责任。文本页已经标记具体来源原文地址,请点击原文查看来源网址,站内文章以及资源内容站长不承诺其正确性,如侵犯了您的权益,请联系站长如有侵权请联系站长,将立刻删除

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信图片_20190322181744_03.jpg

微信扫一扫打赏

请作者喝杯咖啡吧~

支付宝扫一扫领取红包,优惠每天领

二维码1

zhifubaohongbao.png

二维码2

zhifubaohongbao2.png