RabbitMQ(一):Hello World程序

RabbitMQ(一):Hello World程序

内容翻译自:RabbitMQ Tutorials Java版


RabbitMQ(一):Hello World程序

RabbitMQ(二):Work Queues、循环分发、消息确认、持久化、公平分发

RabbitMQ(三):Exchange交换器–fanout

RabbitMQ(四):Exchange交换器–direct

RabbitMQ(五):Exchange交换器–topic

RabbitMQ(六):回调队列callback queue、关联标识correlation id、实现简单的RPC系统

RabbitMQ(七):常用方法说明 与 学习小结


介绍:

RabbitMQ是一个消息代理:它接受并转发消息。你可以把它当成一个邮局:当你想邮寄信件的时候,你会把信件放在投递箱中,并确信邮递员最终会将信件送到收件人的手里。在这个例子中,RabbitMQ就相当与投递箱、邮局和邮递员。

RabbitMQ与邮局的区别在于:RabbitMQ并不处理纸质信件,而是接受、存储并转发二进制数据—消息。

谈到RabbitMQ的消息,通常有几个术语:

(1)生产者:是指发送消息的程序

(2)队列:相当于RabbitMQ的投递箱。尽管消息在RabbitMQ和你的应用之间传递,但是消息仅仅会在队列之中存储。队列只能存储在内存或磁盘中,本质上是一个大的消息缓冲区。不同的生产者可以发送消息到同一个对队列,不同的消费者也可以从同一个队列中获取消息。

(3)消费者:等待接受消息的程序。

注意,生产者、消费者以及RabbitMQ并不一定要在同一个主机上,在绝大部分的应用中它们都不在同一主机上。

在开始教程之前,请确保:你已经安装了RabbitMQ,并且在localhost上运行起来(默认端口5672)。如果你使用了不同的主机或端口,请在下文中的连接设置中
更改相应的参数。


一、Hello World:

在这一部分,我们将会使用Java编写两个小程序:一个发送单个消息的生产者、一个接受消息并打印出消息的消费者。这个消息就是Hello World

下图中,P代表生产者,C代表消费者,中间红色的小箱子就代表队列–RabbitMQ为了让消费者收到消息而保持的消息缓冲区。

RabbitMQ(一):Hello World程序

在这一部分,只需要引入Java客户端依赖即可:amqp-client.jar,也可以通过maven的方式引入:

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>4.1.0</version>
</dependency>

1、生产者:

我们将消息的发布者(生产者)命名为Send,将消息的消费者命名为Recv。发布者将会连接到RabbitMQ,并且发送一条消息,然后退出。

Send.java中,首先引入相关类:

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;

再定义队列的名字:

private final static String QUEUE_NAME = "hello";

然后,创建一个连接到Rabbit服务器的连接:

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

上面的代码中,connection是socket连接的抽象,为我们处理了通信协议版本协商以及认证等。这样,我们就连接到了本地机器上的一个消息代理(broker)。如果想连接到其他机器上的broker,只要修改IP即可。

之后,我们又创建了一个通道(channel),大部分的API操作均在这里完成。

对于Send来说,必须指明消息要发到哪个队列:

channel.queueDeclare(QUEUE_NAME, false, false, false, null);
String message = "Hello World!";
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");

队列的定义是幂等的,它仅仅在不存在时才会创建。消息的内容是一个字节数组,所以你可以随意编码(encode)。

最后,必须将通道和连接关闭。

channel.close();
connection.close();

完整代码

//引入相关Class文件
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Send {

    //定义队列名字
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws Exception {
        //创建连接和通道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        //为通道指明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        String message = "Hello World!";
        
        //发布消息
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
        System.out.println(" [x] Sent '" + message + "'");

        //关闭连接
        channel.close();
        connection.close();
    }
}

2、接收者(消费者):

消费者从RabbitMQ中取出消息。不同于发布者只发送一条消息就退出,这里我们让消费者一直监听消息,并把接受到的消息打印出来。

与Send.java类似,首先引入相关类:

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;

上面引入的DefaultConsumerConsumer接口的实现类,我们使用它来缓冲从服务器push来的消息。
接下来的设置与发布者类似,打开连接和通道,声明我们想消费的队列。注意,这里的队列的名字要与发布者中声明的队列的名字一致。

public class Recv {
  private final static String QUEUE_NAME = "hello";

  public static void main(String[] argv)
      throws java.io.IOException,
             java.lang.InterruptedException {

    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();

    channel.queueDeclare(QUEUE_NAME, fasle, false, false, null);
    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    ...
    }
}

注意,消费者同样声明了队列。这是因为,我们可能在启动生产者之前启动了消费者应用,我们想确保在从一个队列消费消息之前,这个队列是存在的。

接下来,告诉服务器(RabbitMQ)把队列中的消息发过来。因为这个过程是异步的,可以通过DefaultConsumer来进行回调。

Consumer consumer = new DefaultConsumer(channel) {
  @Override
  public void handleDelivery(String consumerTag, Envelope envelope,
                             AMQP.BasicProperties properties, byte[] body)
      throws IOException {
    String message = new String(body, "UTF-8");
    System.out.println(" [x] Received '" + message + "'");
  }
};
channel.basicConsume(QUEUE_NAME, true, consumer);

Consumer的完整代码如下:

package com.maxwell.rabbitdemo;

import com.rabbitmq.client.*;

import java.io.IOException;

public class Recv {

    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws Exception {
        //建立连接和通道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        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");

        //回调消费消息
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
            }
        };
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

这样,消费者就会一直监听声明的队列。运行一次生产者(即Send.java中的main方法),消费者就会打印出接受到的消息。

 


说明:

①与原文略有出入,如有疑问,请参考原文。

②RabbitMQ的官方rabbitmq-tutorials的java示例中,amqp-client版本为3.5,我改为了4.1,否则后续的示例教程中会报错说找不到文件。

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/114609.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • 关于@NotNull 和 @Nullable

    关于@NotNull 和 @Nullable阅读spring源码发现里面用到两个注解@NotNull和@Nullable,现在做一个简单的分析参考文档:避免Java应用中NullPointerException的技巧和最佳实践安卓中的@Nullable和NotNull注释在写程序的时候你可以定义是否可为空指针。通过使用像@NotNull和@Nullable之类的annotation来声明一个方法是否是空指针安全的。…

  • unity update 协程_Unity 协程的原理

    unity update 协程_Unity 协程的原理Unity协程的原理发布时间:2019-06-1318:45,浏览次数:1118,标签:Unity协程不是多线程,协程还是在主线程里面(注:在Unity中非主线程是不可以访问Unity资源的)1、线程、进程和协程的区别进程有自己独立的堆和栈,即不共享堆也不共享栈,进程由操作系统调度线程拥有自己独立的栈和共享的堆,共享堆不共享栈,线程亦有操作系统调度(标准线程是这样的)协程和线程一样共享堆不共…

  • python之路

    python之路

  • airplay影像_播放ftp服务器的视频

    airplay影像_播放ftp服务器的视频http://bbs.weiphone.com/read-htm-tid-1785042.htmliOS客户端软件:AirPlayer[支持RMVB,MKV,MP4,AVI等等格式],另外有个PulgPlayer不支持RMVBMKV等等格式。PC端媒体服务器:WIN7自带的媒体中心(论坛有设置方法的教程,这不介绍了)…

  • R-L模型算法的优缺点_风筝模型公式

    R-L模型算法的优缺点_风筝模型公式介绍Logistic回归算法,名字虽带有回归,但其实是一个分类模型。输出Y=1的对数几率是由输入x的线性函数表示的模型,直接对分类的可能性进行建模,并不是直接对分类的结果(0或者1)进行建模:假设一个样本属于正样本的概率为p,则:LR模型是在线性回归的基础上,把特征进行线性组合,再把组合的结果通过一层sigmoid函数映射成结果是1或是0的概率。逻辑斯蒂回归模型的特点:…

    2022年10月13日
  • vs 注释快捷键_VS2010快捷键

    vs 注释快捷键_VS2010快捷键在VisualStudioIDE中使用快捷键注释代码,无论是行注释还是块注释,第一步一定是选中要注释的内容(取消注释同样要先选中)!当然,如果是行注释,不必选取整行,将光标定位到该行即可!下面整理一下VisualStudioIDE中行注释和块注释的快捷键,以及快捷键的查看与修改。行注释行注释有以下两种方式!方式一:注释:Ctrl+K,Ctrl+C取消:Ctrl+K,Ctrl+U方式二:注释和取消都是这组快捷键:Ctrl+K,Ct…

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号