本站网址
www.kjyfx.com

『Java』RabbitMQ消息队列相关配置及基本使用(六种模式)

一、RabbitMQ是什么?

MQ : 消息总线(Message Queue),是一种跨进程、异步的通信机制,用于上下游传递消息。由消息系统来确保消息的可靠传递。

RabbitMQ是实现了高级消息队列协议(AMQP)的开源消息代理软件(亦称面向消息的中间件)。RabbitMQ服务器是用Erlang语言编写的,它主要的技术特点是可用性,安全性,集群,多协议支持,可视化的客户端,活跃的社区。

RabbitMQ是一种消息中间件,用于处理来自客户端的异步消息。服务端将要发送的消息放入到队列池中。接收端可以根据RabbitMQ配置的转发机制接收服务端发来的消息。RabbitMQ依据指定的转发规则进行消息的转发、缓冲和持久化操作,主要用在多服务器间或单服务器的子系统间进行通信,是分布式系统标准的配置。

 

二、为什么选择RabbitMQ

  • 可靠性(Reliablity):使用了一些机制来保证可靠性,比如持久化、传输确认、发布确认。
  • 灵活的路由(Flexible Routing):在消息进入队列之前,通过Exchange来路由消息。对于典型的路由功能,Rabbit已经提供了一些内置的Exchange来实现。针对更复杂的路由功能,可以将多个Exchange绑定在一起,也通过插件机制实现自己的Exchange。
  • 消息集群(Clustering):多个RabbitMQ服务器可以组成一个集群,形成一个逻辑Broker。
  • 高可用(Highly Avaliable Queues):队列可以在集群中的机器上进行镜像,使得在部分节点出问题的情况下队列仍然可用。
  • 多种协议(Multi-protocol):支持多种消息队列协议,如STOMP、MQTT等。
  • 多种语言客户端(Many Clients):几乎支持所有常用语言,比如Java、.NET、Ruby等。
  • 管理界面(Management UI):提供了易用的用户界面,使得用户可以监控和管理消息Broker的许多方面。
  • 跟踪机制(Tracing):如果消息异常,RabbitMQ提供了消息的跟踪机制,使用者可以找出发生了什么。
  • 插件机制(Plugin System):提供了许多插件,来从多方面进行扩展,也可以编辑自己的插件

 

三、RabbitMQ模型

  • Publisher(消息的生产者):也是一个向交换器发布消息的客户端应用程序。
  • Broker(消息代理) :实际上就是消息服务器实体。
  • VHost(虚拟主机) : 一个消息代理(Broker)里可以开设多个虚拟主机(vhost),用作不同用户的权限分离。
  • Exchange(交换机) : 用来发送消息的AMQP实体,它指定消息按什么路由规则,路由到哪个队列。
  • Binding(绑定) : 它的作用就是把交换机(Exchange)和队列(Queue)按照路由规则绑定起来。
  • Queue(消息队列) :用来保存消息直到发送给消费者。它是消息的容器,也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面,等待消费者连接到这个队列将其取走。
  • Connection(连接) :网络连接,比如一个TCP连接。AMQP连接通常是长连接,Producer和Consumer都是通过TCP连接到RabbitMQ Server的。
  • Channel(通道) :多路复用连接中的一条独立的双向数据流通道。新到是建立在真实的TCP连接内地虚拟链接,AMQP命令都是通过新到发出去的,不管是发布消息、订阅队列还是接收消息,这些动作都是通过信道完成。因为对于操作系统来说,建立和销毁TCP都是非常昂贵的开销,所以引入了信道的概念,以复用一条TCP连接。
  • Consumer(消息的消费者):表示一个从一个消息队列中取得消息的客户端应用程序。
  • Routing Key(路由键) :交换机(Exchange)根据这个关键字进行消息投递,与Queue消息队列进行绑定。
AMQP中消息的路由过程和JMS存在一些差别。AMQP中增加了Exchange和Binging的角色。生产者把消息发布到Exchange上,消息最终到达队列并被消费者接收,而Binding决定交换器的消息应该发送到哪个队列。

 

四、Exchange交换机类型

在RabbitMQ中消息并不会被直接投递到队列中去,而是有生产者将消息发布到交换机中,交换机和一个或多个队列绑定,通过不同的路由规则将消息路由到队列中去,供消费者消费,RabbitMQ中共提供了四种类型交换机(direct、fanout、topic、headers(headers匹配AMQP消息的header而不是路由键(Routing-key),此外headers交换器和direct交换器完全一致,但是性能差了很多,目前几乎用不到了),交换机可以有两个状态:持久(durable)、暂存(transient)。持久化的交换机会在消息代理(broker)重启后依旧存在,而暂存的交换机则不会(它们需要在代理再次上线后重新被声明)。

4.1、直连交换机(Direct)

直连型交换机(direct exchange)是根据消息携带的路由键(routing key)将消息投递给对应队列的。直连交换机要求Publisher和Consumer的路由关键字(routingKey)完全相同才会将消息路由到绑定的队列。直连交换机用来处理消息的单播路由(unicast routing),虽然它可以进行多播。消息中的路由键(routing key)如果和Binding中的binding key一致,交换器就将消息发到对应的队列中。路由键与队列名完全匹配。

 

Publisher:

// 路由关键字
private static final String[] routingKeys = new String[] {"info", "warn", "error",""debug};
// 声明一个交换机并指定它的类型为direct
channel.exchangeDeclare("exchange_direct", "direct")
//发布消息
for (String routingKey : routingKeys) {
   String message = "Send the message : " + severity;
   channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
}

Consumer:

// 路由关键字
private static final String[] routingKeys = new String[] {"info", "warn"};
// 声明一个交换机并指定它的类型为direct
channel.exchangeDeclare("exchange_direct", "direct")
// 声明一个临时队列
String queueName = channel.queueDeclare().getQueue();
// 根据路由关键字绑定队列
for (String routingKey : routingKeys) {
   channel.queueBind(queueName, EXCHANGE_NAME, routingKey);
}

4.2、扇形交换机(Faount)

扇型交换机(funout exchange)会将消息路由给绑定到它身上的所有队列,而不理会绑定的路由键,扇型用来交换机处理消息的广播路由(broadcast routing)。

每个发到fanout类型交换器的消息都会分到所有绑定的队列上去。fanout交换器不处理该路由键,只是简单的将队列绑定到交换器上,每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。很像子网广播,每台子网内的主机都获得了一份复制的消息。fanout类型转发消息是最快的。

Publisher:

private static final String EXCHANGE_NAME = "exchange.fanout";
// 在发布端可以不声明队列
//channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
for(int i=0;i<5;i++){
   String message="this is number"+i;
   // 路由关键字不能为null,填写""
   channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes());
   System.out.println(" [x] sent ' " + message + " '");
}

Consumer:

 private static final String EXCHANGE_NAME = "exchange.fanout";
 private final static String QUEUE_NAME = "queue.fanout";
 channel.queueDeclare(QUEUE_NAME,true,false,false,null);
 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
 // 绑定队列
 channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"");

 

4.3、主题交换机(Topic)

主题交换机(topic exchanges)通过对消息的路由键和队列到交换机的绑定模式之间的匹配,将消息路由给一个或多个队列,属于多播路。topic可以进行模糊匹配,可以使用星号和井号#这两个通配符来进行模糊匹配,其中 号可以代替一个单词 # 号可以代替任意个单词,但是需要注意的是topic交换机的路由键也不是可以随意设置的,必须是由点隔开的一系列的标识符组成。标识符一般和消息的某些特性相关,可以定义任意数量的标识符,上限为255个字节,当路由键可以模糊匹配上的时候就能将消息映射到绑定的队列中去。

topic交换器通过模式匹配分配消息的路由键属性,将路由键和某个模式进行匹配,此时队列需要绑定到一个模式上。它将路由键(routing-key)和绑定键(bingding-key)的字符串切分成单词,这些单词之间用点隔开。它同样也会识别两个通配符:”#”和”*”。#匹配0个或多个单词,匹配不多不少一个单词。

 

默认交换机(Default)

默认交换机(default exchange)不是一个真正的交换机类型,实际上是一个由消息代理(Broker)预先声明好的没有名字(名字为空字符串)的直连交换机。它有一个特殊的属性:那就是每个新建队列(queue)都会自动绑定到默认交换机上,绑定的路由键(routing key)名称与队列名称相同。
很多时候我们对于一些不复杂的场景都会使用这一特殊属性。

Publisher:

private static final String QUEUE_NAME="task_queue"; 
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());

Consumer:

private static final String QUEUE_NAME="task_queue"; 
channel.queueDeclare(QUEUE_NAME, false, false, false, null);

上面的伪代码看起来我们并没有使用交换机,而是直接将消息投递到了队列中去,但实际上这个队列被绑定到了默认交换机上,而路由键就是队列名称。

 

五、使用场景

  • 上下游逻辑解耦&&物理解耦
  • 保证数据最终一致性
  • 广播
  • 错峰流控等等

5.1、场景 一(服务解耦)

假设有这样一个场景, 服务A产生数据, 而服务B,C,D需要这些数据, 那么我们可以在A服务中直接调用B,C,D服务,把数据传递到下游服务即可。但是,随着我们的应用规模不断扩大,会有更多的服务需要A的数据,如果有几十甚至几百个下游服务,而且会不断变更,再加上还要考虑下游服务出错的情况,那么A服务中调用代码的维护会极为困难。这是由于服务之间耦合度过于紧密

 

再来考虑用RabbitMQ解耦的情况

A服务只需要向消息服务器发送消息,而不用考虑谁需要这些数据;下游服务如果需要数据,自行从消息服务器订阅消息,不再需要数据时则取消订阅即可

5.2、场景二(流量削峰)

假设我们有一个应用,平时访问量是每秒300请求,我们用一台服务器即可轻松应对

而在高峰期,访问量瞬间翻了十倍,达到每秒3000次请求,那么单台服务器肯定无法应对,这时我们可以考虑增加到10台服务器,来分散访问压力

但如果这种瞬时高峰的情况每天只出现一次,每次只有半小时,那么我们10台服务器在多数时间都只分担每秒几十次请求,这样就有点浪费资源了

这种情况,我们就可以使用RabbitMQ来进行流量削峰,高峰情况下,瞬间出现的大量请求数据,先发送到消息队列服务器,排队等待被处理,而我们的应用,可以慢慢的从消息队列接收请求数据进行处理,这样把数据处理时间拉长,以减轻瞬时压力。这是消息队列服务器非常典型的应用场景

 

5.3、场景三(异步调用)

考虑定外卖支付成功的情况

支付后要发送支付成功的通知,再寻找外卖小哥来进行配送,而寻找外卖小哥的过程非常耗时,尤其是高峰期,可能要等待几十秒甚至更长

这样就造成整条调用链路响应非常缓慢

而如果我们引入RabbitMQ消息队列,订单数据可以发送到消息队列服务器,那么调用链路也就可以到此结束,订单系统则可以立即得到响应,整条链路的响应时间只有200毫秒左右

寻找外卖小哥的应用可以以异步的方式从消息队列接收订单消息,再执行耗时的寻找操作

六、RabbitMQ安装与设置

RabbitMQ使用了Erlang开发语言,Erlang是为电话交换机开发的语言,天生自带高并发光环,和高可用特性

RabbitMQ安装时所需的包:https://www.lanzous.com/i766fqf

 

6.1、安装Erlang语言包(rabbitmq官方精简的Erlang语言包

下载地址 : https://github.com/rabbitmq/erlang-rpm

(在线下载速度较慢)

# 下载Erlang语言包
wget https://github.com/rabbitmq/erlang-rpm/releases/download/v21.2.6/erlang-21.2.6-1.el7.x86_64.rpm

# 安装Erlang
rpm -ivh erlang-21.2.6-1.el7.x86_64.rpm --force --nodeps

 

6.2、安装socat依赖

下载地址:https://pkgs.org/download/socat

(在线下载速度较慢)

# 下载 socat rpm
wget http://mirror.centos.org/centos/7/os/x86_64/Packages/socat-1.7.3.2-2.el7.x86_64.rpm

# 安装 socat 依赖包
rpm -ivh socat-1.7.3.2-2.el7.x86_64.rpm

 

6.3、安装RabbitMQ

下载地址http://www.rabbitmq.com/install-rpm.html#downloads

(在线下载速度较慢)

# 下载 rpm 包
wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.7.13/rabbitmq-server-3.7.13-1.el7.noarch.rpm

# 安装 rpm 包
rpm -ivh rabbitmq-server-3.7.13-1.el7.noarch.rpm

 

6.4、rabbitmq启动和停止命令

# 设置服务,开机自动启动
chkconfig rabbitmq-server on

# 启动服务
service rabbitmq-server start

# 停止服务
service rabbitmq-server stop

 

6.5、rabbitmq管理界面

# 开启管理界面插件
rabbitmq-plugins enable rabbitmq_management

# 防火墙打开 15672 管理端口
firewall-cmd --zone=public --add-port=15672/tcp --permanent
firewall-cmd --reload

 

6.6、添加用户(用户名与密码都是是admin)

# 添加用户
rabbitmqctl add_user admin admin

# 新用户设置用户为超级管理员
rabbitmqctl set_user_tags admin administrator

访问服务器的15672端口,例如 : http://192.168.71.128:15672

出现可视化管理界面

6.7、设置访问权限

用户管理参考:https://www.cnblogs.com/AloneSword/p/4200051.html

 

6.8、开放客户端连接端口

# 打开客户端连接端口
firewall-cmd --zone=public --add-port=5672/tcp --permanent
firewall-cmd --reload

主要端口介绍

  • 4369 — erlang发现口
  • 5672 — client端通信口
  • 15672 — 管理界面ui端口
  • 25672 — server间内部通信口

 

七、RabbitMQ六种工作模式

pom文件首先添加 slf4j 依赖, 和 rabbitmq amqp 依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.kjyfx</groupId>
    <artifactId>RabbitMQ</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.4.3</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.8.0-alpha2</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.8.0-alpha2</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

7.1、简单模式

RabbitMQ是一个消息中间件,你可以想象它是一个邮局。当你把信件放到邮箱里时,能够确信邮递员会正确地递送你的信件。RabbitMq就是一个邮箱、一个邮局和一个邮递员。

  • 发送消息的程序是生产者
  • 队列就代表一个邮箱。虽然消息会流经RbbitMQ和你的应用程序,但消息只能被存储在队列里。队列存储空间只受服务器内存和磁盘限制,它本质上是一个大的消息缓冲区。多个生产者可以向同一个队列发送消息,多个消费者也可以从同一个队列接收消息.
  • 消费者等待从队列接收消息

生产者发送消息:Provider

package rabbitmq.simple;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
 *RabbitMQ简单模式生产者
 */
public class Provider {
    public static void main(String[] args) throws Exception {
        //创建连接工厂,并设置连接信息
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.71.128");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");

        /*
         * 与rabbitmq服务器建立连接,
         * rabbitmq服务器端使用的是nio,会复用tcp连接,
         * 并开辟多个信道与客户端通信
         * 以减轻服务器端建立连接的开销
         */
        Connection c = factory.newConnection();

        //建立信道
        Channel ch = c.createChannel();

        /*
         * 声明队列,会在rabbitmq中创建一个队列
         * 如果已经创建过该队列,就不能再使用其他参数来创建
         *
         * 参数含义:
         *   -queue: 队列名称
         *   -durable: 队列持久化,true表示RabbitMQ重启后队列仍存在
         *   -exclusive: 排他,true表示限制仅当前连接可用
         *   -autoDelete: 当最后一个消费者断开后,是否删除队列
         *   -arguments: 其他参数
         */
        ch.queueDeclare("helloworld",false,false,false,null);
        /*
         * 发布消息
         * 这里把消息向默认交换机发送.
         * 默认交换机隐含与所有队列绑定,routing key即为队列名称
         *
         * 参数含义:
         *     -exchange: 交换机名称,空串表示默认交换机"(AMQP default)",不能用 null
         *     -routingKey: 对于默认交换机,路由键就是目标队列名称
         *     -props: 其他参数,例如头信息
         *     -body: 消息内容byte[]数组
         */
        ch.basicPublish("","helloworld",null,"rabbit is run".getBytes());
        System.out.println("消息已发出");
    }
}

 

消费者接收消息:Consumer

package rabbitmq.simple;

import com.rabbitmq.client.*;
import java.io.IOException;
/**
 *RabbitMQ简单模式消费者
 */
public class Consumer {
    public static void main(String[] args) throws Exception {
        //连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.71.128");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");

        //建立连接
        Connection c = factory.newConnection();
        //建立信道
        Channel ch = c.createChannel();

        //声明队列,如果该队列已经创建过,则不会重复创建
        ch.queueDeclare("helloworld",false,false,false,null);
        System.out.println("等待接收数据");

        //收到消息后用来处理消息的回调对象
        DeliverCallback callback = new DeliverCallback() {
            @Override
            public void handle(String s, Delivery message) throws IOException {
                String msg = new String(message.getBody(),"UTF-8");
                System.out.println("收到信息:"+msg);
            }
        };

        //消费者取消时的回调对象
        CancelCallback cancel = new CancelCallback() {
            @Override
            public void handle(String s) throws IOException {

            }
        };

        ch.basicConsume("helloworld",true,callback,cancel);
    }
}

 

7.2、工作模式

工作队列(即任务队列)背后的主要思想是避免立即执行资源密集型任务,并且必须等待它完成。相反,我们将任务安排在稍后完成。

我们将任务封装为消息并将其发送到队列。后台运行的工作进程将获取任务并最终执行任务。当运行多个消费者时,任务将在它们之间分发。

使用任务队列的一个优点是能够轻松地并行工作。如果我们正在积压工作任务,我们可以添加更多工作进程,这样就可以轻松扩展。

 

生产者发送消息(功能解析,慎用):Provider

package rabbitmq.work;

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

import java.util.Scanner;

/**
 * RabbitMQ工作模式生产者
 */
public class Provider {
    public static void main(String[] args) throws Exception {
        //创建连接工厂,并设置连接信息
        ConnectionFactory f = new ConnectionFactory();
        f.setHost("192.168.71.128");
        f.setPort(5672);
        f.setUsername("admin");
        f.setPassword("admin");

        //从连接工厂建立新的连接
        Connection c = f.newConnection();

        //从连接创建信道
        Channel ch = c.createChannel();
        //声明队列
        ch.queueDeclare("helloworld",false,false,false,null);

        while (true){
            System.out.println("请输入消息:");
            String msg = new Scanner(System.in).nextLine();
            if ("exit".equals(msg)){
                break;
            }
            //参数:exchage,routingKey,props,body
            ch.basicPublish("","helloworld",null,msg.getBytes());
            System.out.println("已发送消息:"+msg);
        }
        c.close();
    }
}

 

消费者接收消息(手动消息确认默认开启,数据易丢失,慎用):Consumer

package rabbitmq.work;

import com.rabbitmq.client.*;
import java.io.IOException;

/**
 * RabbitMQ工作模式消费者(手动消息确认默认开启,当rabbitmq宕机后会造成数据丢失)
 */
public class Consumer {

    public static void main(String[] args) throws Exception {
        //创建连接工厂,并设置连接信息
        ConnectionFactory f = new ConnectionFactory();
        f.setHost("192.168.71.128");
        f.setPort(5672);
        f.setUsername("admin");
        f.setPassword("admin");

        Connection c = f.newConnection();
        Channel ch = c.createChannel();

        ch.queueDeclare("helloworld",false,false,false,null);
        System.out.println("等待接收数据");

        //收到消息后用来处理消息的回调对象
        DeliverCallback callback = new DeliverCallback() {
            @Override
            public void handle(String s, Delivery message) throws IOException {
                String msg = new String(message.getBody(),"utf-8");
                System.out.println("收到消息:"+msg);

                //遍历字符串中的字符,每个点使进程暂停一秒
                for (int i = 0; i <msg.length() ; i++) {
                    if(msg.charAt(i) == '.'){
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
                System.out.println("处理结束");
            }
        };

        //消费者取消时的回调对象
        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String s) throws IOException {

            }
        };
        ch.basicConsume("helloworld",true,callback,cancelCallback);
    }
}

 

运行测试 : 一个生产者,两个消费者

生产者发送多条消息, 如: 1,2,3,4,5. 两个消费者分别收到:

  • 消费者一 : 1,3,5
  • 消费者二 : 2,4

rabbitmq在所有消费者中轮询分发消息,把消息均匀地发送给所有消费者

 

消息确认

一个消费者接收消息后,在消息没有完全处理完时就挂掉了,那么这时会发生什么呢?

就现在的代码来说,rabbitmq把消息发送给消费者后,会立即删除消息,那么消费者挂掉后,它没来得及处理的消息就会丢失

如果生产者发送以下消息:
1…….
2
3
4
5

两个消费者分别收到:

  • 消费者一: 1……., 3, 5
  • 消费者二: 2, 4

当消费者一收到所有消息后,要话费7秒时间来处理第一条消息,这期间如果关闭该消费者,那么1未处理完成,3,5则没有被处理

我们并不想丢失任何消息, 如果一个消费者挂掉,我们想把它的任务消息派发给其他消费者

为了确保消息不会丢失,rabbitmq支持消息确认(回执)。当一个消息被消费者接收到并且执行完成后,消费者会发送一个ack (acknowledgment) 给rabbitmq服务器, 告诉他我已经执行完成了,你可以把这条消息删除了。

如果一个消费者没有返回消息确认就挂掉了(信道关闭,连接关闭或者TCP链接丢失),rabbitmq就会明白,这个消息没有被处理完成,rebbitmq就会把这条消息重新放入队列,如果在这时有其他的消费者在线,那么rabbitmq就会迅速的把这条消息传递给其他的消费者,这样就确保了没有消息会丢失。

这里不存在消息超时, rabbitmq只在消费者挂掉时重新分派消息, 即使消费者花非常久的时间来处理消息也可以。

手动消息确认默认是开启的,前面的例子我们通过autoAck=ture把它关闭了。我们现在要把它设置为false,然后工作进程处理完意向任务时,发送一个消息确认(回执)。

消费者接收消息(手动消息确认关闭,数据不易丢失,功能解析,慎用):Consumer

使用以下代码,就算杀掉一个正在处理消息的工作进程也不会丢失任何消息,工作进程挂掉之后,没有确认的消息就会被自动重新传递。

package rabbitmq.work;

import com.rabbitmq.client.*;
import java.io.IOException;

/**
 * RabbitMQ工作模式消费者(关闭手动确认,失败后会把消息传递给其他消费者,数据不会丢失)
 */
public class Consumer {

    public static void main(String[] args) throws Exception {
        //创建连接工厂,并设置连接信息
        ConnectionFactory f = new ConnectionFactory();
        f.setHost("192.168.71.128");
        f.setPort(5672);
        f.setUsername("admin");
        f.setPassword("admin");

        Connection c = f.newConnection();
        Channel ch = c.createChannel();

        ch.queueDeclare("helloworld",false,false,false,null);
        System.out.println("等待接收数据");

        //收到消息后用来处理消息的回调对象
        DeliverCallback callback = new DeliverCallback() {
            @Override
            public void handle(String s, Delivery message) throws IOException {
                String msg = new String(message.getBody(),"utf-8");
                System.out.println("收到消息:"+msg);

                //遍历字符串中的字符,每个点使进程暂停一秒
                for (int i = 0; i <msg.length() ; i++) {
                    if(msg.charAt(i) == '.'){
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
                System.out.println("处理结束");
                //发送回执
                ch.basicAck(message.getEnvelope().getDeliveryTag(), false);
            }
        };

        //消费者取消时的回调对象
        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String s) throws IOException {

            }
        };
        
        //ch.basicConsume("helloworld",true,callback,cancelCallback);
        //autoAck设置为false,则需要手动确认发送回执
        ch.basicConsume("helloworld", false, callback, cancelCallback);
    }

}

忘记确认(ack)是一个常见的错误, 这样后果是很严重的, 由于未确认的消息不会被释放, rabbitmq会吃掉越来越多的内存

可以使用下面命令打印工作队列中未确认消息的数量

rabbitmqctl list_queues name messages_ready messages_unacknowledged

 

消息持久化

当rabbitmq关闭时, 我们队列中的消息仍然会丢失, 除非明确要求它不要丢失数据

要求rabbitmq不丢失数据要做如下两点: 把队列和消息都设置为可持久化(durable)

队列设置为可持久化, 可以在定义队列时指定参数durable为true

//第二个参数是持久化参数durable
ch.queueDeclare("helloworld", true, false, false, null);

 

我们已经定义过队列”helloworld”是不可持久化的, 对已存在的队列, rabbitmq不允许对其定义不同的参数, 否则会出错,

 

所以这里我们定义一个不同名字的队列“task_queue”,修改持久化参数为true,生产者和消费者代码都要修改。这样即使rabbitmq重新启动, 队列也不会丢失

//定义一个新的队列,名为 task_queue
//第二个参数是持久化参数 durable
ch.queueDeclare("task_queue", true, false, false, null);

 

现在我们再设置队列中消息的持久化, 使用MessageProperties.PERSISTENT_TEXT_PLAIN参数

//第三个参数设置消息持久化
ch.basicPublish("", "task_queue",
 MessageProperties.PERSISTENT_TEXT_PLAIN,
 msg.getBytes());

 

合理地分发

rabbitmq会一次把多个消息分发给消费者, 这样可能造成有的消费者非常繁忙, 而其它消费者空闲. 而rabbitmq对此一无所知, 仍然会均匀的分发消息

我们可以使用 basicQos(1) 方法, 这告诉rabbitmq一次只向消费者发送一条消息, 在返回确认回执前, 不要向消费者发送新消息. 而是把消息发给下一个空闲的消费者

 

下面是”工作模式”最终完成的生产者和消费者代码

生产者发送消息(最终版代码,持久,合理分发):Provider

package rabbitmq.work;

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

import java.util.Scanner;

/**
 * RabbitMQ工作模式生产者(持久、合理分发)
 */
public class Provider {
    public static void main(String[] args) throws Exception {
        ConnectionFactory f = new ConnectionFactory();
        f.setHost("192.168.71.128");
        f.setPort(5672);
        f.setUsername("admin");
        f.setPassword("admin");

        Connection c = f.newConnection();
        Channel ch = c.createChannel();

        //第二个参数设置队列持久化
        ch.queueDeclare("task_queue", true,false,false,null);

        while (true) {
            System.out.print("输入消息: ");
            String msg = new Scanner(System.in).nextLine();
            if ("exit".equals(msg)) {
                break;
            }

            //第三个参数设置消息持久化
            ch.basicPublish("", "task_queue", MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes("UTF-8"));
            System.out.println("消息已发送: "+msg);
        }

        c.close();
    }

}

 

消费者接收消息(最终版代码,持久,合理分发):Consumer

package rabbitmq.work;

import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * RabbitMQ工作模式消费者(持久、合理分发)
 */
public class Consumer {

    public static void main(String[] args) throws Exception {
        ConnectionFactory f = new ConnectionFactory();
        f.setHost("192.168.64.140");
        f.setUsername("admin");
        f.setPassword("admin");
        Connection c = f.newConnection();
        Channel ch = c.createChannel();

        //第二个参数设置队列持久化
        ch.queueDeclare("task_queue",true,false,false,null);

        System.out.println("等待接收数据");

        ch.basicQos(1); //一次只接收一条消息

        //收到消息后用来处理消息的回调对象
        DeliverCallback callback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                String msg = new String(message.getBody(), "UTF-8");
                System.out.println("收到: "+msg);
                for (int i = 0; i < msg.length(); i++) {
                    if (msg.charAt(i)=='.') {
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                        }
                    }
                }
                System.out.println("处理结束");
                //发送回执
                ch.basicAck(message.getEnvelope().getDeliveryTag(), false);
            }
        };

        //消费者取消时的回调对象
        CancelCallback cancel = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
            }
        };

        //autoAck设置为false,则需要手动确认发送回执
        ch.basicConsume("task_queue", false, callback, cancel);
    }
}

 

7.3、发布订阅模式

 

在前面的例子中,我们任务消息只交付给一个工作进程。在这部分,我们将向多个消费者传递同一条消息。这种模式称为“发布/订阅”。

为了说明该模式,我们将构建一个简单的日志系统。它将由两个程序组成——第一个程序将发出日志消息,第二个程序接收它们。

在我们的日志系统中,接收程序的每个运行副本都将获得消息。这样,我们就可以运行一个消费者并将日志保存到磁盘; 同时我们可以运行另一个消费者在屏幕上打印日志。

最终, 消息会被广播到所有消息接受者

 

Exchanges 交换机

 

RabbitMQ消息传递模型的核心思想是,生产者永远不会将任何消息直接发送到队列。实际上,通常生产者甚至不知道消息是否会被传递到任何队列。

相反,生产者只能向交换机(Exchange)发送消息。交换机是一个非常简单的东西。一边接收来自生产者的消息,另一边将消息推送到队列。交换器必须确切地知道如何处理它接收到的消息。它应该被添加到一个特定的队列中吗?它应该添加到多个队列中吗?或者它应该被丢弃。这些规则由exchange的类型定义。

有几种可用的交换类型:direct、topic、header和fanout。我们将关注最后一个——fanout。让我们创建一个这种类型的交换机,并称之为 logs: ch.exchangeDeclare(“logs”, “fanout”);

fanout交换机非常简单。它只是将接收到的所有消息广播给它所知道的所有队列。这正是我们的日志系统所需要的。

我们前面使用的队列具有特定的名称(还记得hello和task_queue吗?)能够为队列命名对我们来说至关重要——我们需要将工作进程指向同一个队列,在生产者和消费者之间共享队列。

但日志记录案例不是这种情况。我们想要接收所有的日志消息,而不仅仅是其中的一部分。我们还只对当前的最新消息感兴趣,而不是旧消息。

要解决这个问题,我们需要两件事。首先,每当我们连接到Rabbitmq时,我们需要一个新的空队列。为此,我们可以创建一个具有随机名称的队列,或者,更好的方法是让服务器为我们选择一个随机队列名称。其次,一旦断开与使用者的连接,队列就会自动删除。在Java客户端中,当我们不向queueDeclare()提供任何参数时,会创建一个具有生成名称的、非持久的、独占的、自动删除队列

//自动生成队列名
//非持久,独占,自动删除
String queueName = ch.queueDeclare().getQueue();

 

绑定 Bindings

我们已经创建了一个fanout交换机和一个队列。现在我们需要告诉exchange向指定队列发送消息。exchange和队列之间的关系称为绑定。

//指定的队列,与指定的交换机关联起来
//成为绑定 -- binding
//第三个参数时 routingKey, 由于是fanout交换机, 这里忽略 routingKey
ch.queueBind(queueName, "logs", "");

现在, logs交换机将会向我们指定的队列添加消息

列出绑定关系指令:rabbitmqctl list_bindings

 

生产者发送消息:Provider

package rabbitmq.subscribe;

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

import java.util.Scanner;

/**
 * RabbitMQ发布订阅模式生产者
 */
public class Provider {
    public static void main(String[] args) throws Exception {
        ConnectionFactory f = new ConnectionFactory();
        f.setHost("192.168.71.128");
        f.setPort(5672);
        f.setUsername("admin");
        f.setPassword("admin");

        Connection c = f.newConnection();
        Channel ch = c.createChannel();

        //定义名字为logs的交换机,交换机类型为fanout
        //这一步是必须的,因为禁止发布到不存在的交换。
        ch.exchangeDeclare("logs", "fanout");

        while (true) {
            System.out.print("输入消息: ");
            String msg = new Scanner(System.in).nextLine();
            if ("exit".equals(msg)) {
                break;
            }

            //第一个参数,向指定的交换机发送消息
            //第二个参数,不指定队列,由消费者向交换机绑定队列
            //如果还没有队列绑定到交换器,消息就会丢失,
            //但这对我们来说没有问题;即使没有消费者接收,我们也可以安全地丢弃这些信息。
            ch.basicPublish("logs", "", null, msg.getBytes("UTF-8"));
            System.out.println("消息已发送: "+msg);
        }
        c.close();
    }
}

 

消费者接收消息:Consumer

package rabbitmq.subscribe;

import com.rabbitmq.client.*;
import java.io.IOException;

/**
 * RabbitMQ发布订阅模式消费者
 */
public class Consumer {

    public static void main(String[] args) throws Exception {
        ConnectionFactory f = new ConnectionFactory();
        f.setHost("192.168.71.128");
        f.setUsername("admin");
        f.setPassword("admin");
        Connection c = f.newConnection();
        Channel ch = c.createChannel();

        //定义名字为 logs 的交换机, 它的类型是 fanout
        ch.exchangeDeclare("logs", "fanout");

        //自动生成对列名,
        //非持久,独占,自动删除
        String queueName = ch.queueDeclare().getQueue();

        //把该队列,绑定到 logs 交换机
        //对于 fanout 类型的交换机, routingKey会被忽略,不允许null值
        ch.queueBind(queueName, "logs", "");

        System.out.println("等待接收数据");

        //收到消息后用来处理消息的回调对象
        DeliverCallback callback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                String msg = new String(message.getBody(), "UTF-8");
                System.out.println("收到: "+msg);
            }
        };

        //消费者取消时的回调对象
        CancelCallback cancel = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
            }
        };

        ch.basicConsume(queueName, true, callback, cancel);
    }

}

 

 

7.4、路由模式

 

在上一小节,我们构建了一个简单的日志系统。我们能够向多个接收者广播日志消息。

在这一节,我们将向其添加一个特性—我们将只订阅所有消息中的一部分。例如,我们只接收关键错误消息并保存到日志文件(以节省磁盘空间),同时仍然能够在控制台上打印所有日志消息。

 

绑定 Bindings

在上一节,我们已经创建了队列与交换机的绑定。使用下面这样的代码:

ch.queueBind(queueName, "logs", "");

绑定是交换机和队列之间的关系。这可以简单地理解为:队列对来自此交换的消息感兴趣。

绑定可以使用额外的routingKey参数。为了避免与basic_publish参数混淆,我们将其称为bindingKey。这是我们如何创建一个键绑定:

ch.queueBind(queueName, EXCHANGE_NAME, "black");

bindingKey的含义取决于交换机类型。我们前面使用的fanout交换机完全忽略它。

 

直连交换机 Direct exchange

上一节中的日志系统向所有消费者广播所有消息。我们希望扩展它,允许根据消息的严重性过滤消息。例如,我们希望将日志消息写入磁盘的程序只接收关键error,而不是在warning或info日志消息上浪费磁盘空间。

前面我们使用的是fanout交换机,这并没有给我们太多的灵活性——它只能进行简单的广播。

我们将用直连交换机(Direct exchange)代替。它背后的路由算法很简单——消息传递到bindingKey与routingKey完全匹配的队列。为了说明这一点,请考虑以下设置

其中我们可以看到直连交换机X,它绑定了两个队列。第一个队列用绑定键orange绑定,第二个队列有两个绑定,一个绑定black,另一个绑定键green。

这样设置,使用路由键orange发布到交换器的消息将被路由到队列Q1。带有black或green路由键的消息将转到Q2。而所有其他消息都将被丢弃。

 

多重绑定 Multiple bindings

使用相同的bindingKey绑定多个队列是完全允许的。如图所示,可以使用binding key black将X与Q1和Q2绑定。在这种情况下,直连交换机的行为类似于fanout,并将消息广播给所有匹配的队列。一条路由键为black的消息将同时发送到Q1和Q2。

 

发送日志

我们将在日志系统中使用这个模型。我们把消息发送到一个Direct交换机,而不是fanout。我们将提供日志级别作为routingKey。这样,接收程序将能够选择它希望接收的级别。让我们首先来看发出日志。

和前面一样,我们首先需要创建一个exchange:

//参数1: 交换机名
//参数2: 交换机类型
ch.exchangeDeclare("direct_logs", "direct");

接着来看发送消息的代码

//参数1: 交换机名
//参数2: routingKey, 路由键,这里我们用日志级别,如"error","info","warning"
//参数3: 其他配置属性
//参数4: 发布的消息数据 
ch.basicPublish("direct_logs", "error", null, message.getBytes());

 

订 阅

接收消息的工作原理与前面一样,但有一个例外——我们将为感兴趣的每个日志级别创建一个新的绑定, 示例代码如下:

ch.queueBind(queueName, "logs", "info");
ch.queueBind(queueName, "logs", "warning");

 

生产者发送消息:Provider

package rabbitmq.routing;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.Random;
import java.util.Scanner;

/**
 * RabbitMQ路由模式生产者
 */
public class Provider {

    public static void main(String[] args) throws Exception {
        String[] a = {"warning", "info", "error"};

        ConnectionFactory f = new ConnectionFactory();
        f.setHost("192.168.71.128");
        f.setPort(5672);
        f.setUsername("admin");
        f.setPassword("admin");

        Connection c = f.newConnection();
        Channel ch = c.createChannel();

        //参数1: 交换机名
        //参数2: 交换机类型
        ch.exchangeDeclare("direct_logs", BuiltinExchangeType.DIRECT);

        while (true) {
            System.out.print("输入消息: ");
            String msg = new Scanner(System.in).nextLine();
            if ("exit".equals(msg)) {
                break;
            }

            //随机产生日志级别
            String level = a[new Random().nextInt(a.length)];

            //参数1: 交换机名
            //参数2: routingKey, 路由键,这里我们用日志级别,如"error","info","warning"
            //参数3: 其他配置属性
            //参数4: 发布的消息数据
            ch.basicPublish("direct_logs", level, null, msg.getBytes());
            System.out.println("消息已发送: "+level+" - "+msg);

        }
        c.close();
    }
}

 

消费者接收消息:Consumer

package rabbitmq.routing;

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.Scanner;

/**
 * RabbitMQ路由模式消费者
 */
public class Consumer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory f = new ConnectionFactory();
        f.setHost("192.168.71.128");
        f.setUsername("admin");
        f.setPassword("admin");
        Connection c = f.newConnection();
        Channel ch = c.createChannel();

        //定义名字为 direct_logs 的交换机, 它的类型是 "direct"
        ch.exchangeDeclare("direct_logs", BuiltinExchangeType.DIRECT);

        //自动生成对列名,
        //非持久,独占,自动删除
        String queueName = ch.queueDeclare().getQueue();

        System.out.println("输入接收的日志级别,用空格隔开:");
        String[] a = new Scanner(System.in).nextLine().split("\\s");

        //把该队列,绑定到 direct_logs 交换机
        //允许使用多个 bindingKey
        for (String level : a) {
            ch.queueBind(queueName, "direct_logs", level);
        }

        System.out.println("等待接收数据");

        //收到消息后用来处理消息的回调对象
        DeliverCallback callback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                String msg = new String(message.getBody(), "UTF-8");
                String routingKey = message.getEnvelope().getRoutingKey();
                System.out.println("收到: "+routingKey+" - "+msg);
            }
        };

        //消费者取消时的回调对象
        CancelCallback cancel = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
            }
        };
        ch.basicConsume(queueName, true, callback, cancel);
    }
}

 

7.5、主题模式

在上一小节,我们改进了日志系统。我们没有使用只能进行广播的fanout交换机,而是使用Direct交换机,从而可以选择性接收日志。

虽然使用Direct交换机改进了我们的系统,但它仍然有局限性——它不能基于多个标准进行路由。

在我们的日志系统中,我们可能不仅希望根据级别订阅日志,还希望根据发出日志的源订阅日志。

这将给我们带来很大的灵活性——我们可能只想接收来自“cron”的关键错误,但也要接收来自“kern”的所有日志。

要在日志系统中实现这一点,我们需要了解更复杂的Topic交换机。

主题交换机 Topic exchange

发送到Topic交换机的消息,它的的routingKey,必须是由点分隔的多个单词。单词可以是任何东西,但通常是与消息相关的一些特性。几个有效的routingKey示例:“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”。routingKey可以有任意多的单词,最多255个字节。

bindingKey也必须采用相同的形式。Topic交换机的逻辑与直连交换机类似——使用特定routingKey发送的消息将被传递到所有使用匹配bindingKey绑定的队列。bindingKey有两个重要的特殊点:

  • * 可以通配单个单词。
  • # 可以通配零个或多个单词。

用一个例子来解释这个问题是最简单的

在本例中,我们将发送描述动物的消息。这些消息将使用由三个单词(两个点)组成的routingKey发送。routingKey中的第一个单词表示速度,第二个是颜色,第三个是物种:“<速度>.<颜色>.<物种>”。

我们创建三个绑定:Q1与bindingKey “*.orange.*” 绑定。和Q2是 “*.*.rabbit” 和 “lazy.#” 。

这些绑定可概括为:

Q1对所有橙色的动物感兴趣。
Q2想接收关于兔子和慢速动物的所有消息。
将routingKey设置为”quick.orange.rabbit”的消息将被发送到两个队列。消息 “lazy.orange.elephant”也发送到它们两个。另外”quick.orange.fox”只会发到第一个队列,”lazy.brown.fox”只发给第二个。”lazy.pink.rabbit”将只被传递到第二个队列一次,即使它匹配两个绑定。”quick.brown.fox”不匹配任何绑定,因此将被丢弃。

如果我们违反约定,发送一个或四个单词的信息,比如”orange”或”quick.orange.male.rabbit”,会发生什么?这些消息将不匹配任何绑定,并将丢失。

另外,”lazy.orange.male.rabbit”,即使它有四个单词,也将匹配最后一个绑定,并将被传递到第二个队列。

 

生产者发送消息:Provider

package rabbitmq.topic;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.Scanner;

/**
 * RabbitMQ主题模式生产者
 */
public class Provider {

    public static void main(String[] args) throws Exception {
        ConnectionFactory f = new ConnectionFactory();
        f.setHost("192.168.71.128");
        f.setPort(5672);
        f.setUsername("admin");
        f.setPassword("admin");

        Connection c = f.newConnection();
        Channel ch = c.createChannel();

        //参数1: 交换机名
        //参数2: 交换机类型
        ch.exchangeDeclare("topic_logs", BuiltinExchangeType.TOPIC);

        while (true) {
            System.out.print("输入消息: ");
            String msg = new Scanner(System.in).nextLine();
            if ("exit".contentEquals(msg)) {
                break;
            }
            System.out.print("输入routingKey: ");
            String routingKey = new Scanner(System.in).nextLine();

            //参数1: 交换机名
            //参数2: routingKey, 路由键,这里我们用日志级别,如"error","info","warning"
            //参数3: 其他配置属性
            //参数4: 发布的消息数据
            ch.basicPublish("topic_logs", routingKey, null, msg.getBytes());

            System.out.println("消息已发送: "+routingKey+" - "+msg);
        }
        c.close();
    }
}

 

消费者接收消息:Consumer

package rabbitmq.topic;

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.Scanner;

/**
 * RabbitMQ主题模式消费者
 */
public class Consumer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory f = new ConnectionFactory();
        f.setHost("192.168.71.128");
        f.setUsername("admin");
        f.setPassword("admin");
        Connection c = f.newConnection();
        Channel ch = c.createChannel();

        ch.exchangeDeclare("topic_logs", BuiltinExchangeType.TOPIC);

        //自动生成对列名,
        //非持久,独占,自动删除
        String queueName = ch.queueDeclare().getQueue();

        System.out.println("输入bindingKey,用空格隔开:");
        String[] a = new Scanner(System.in).nextLine().split("\\s");

        //把该队列,绑定到 topic_logs 交换机
        //允许使用多个 bindingKey
        for (String bindingKey : a) {
            ch.queueBind(queueName, "topic_logs", bindingKey);
        }

        System.out.println("等待接收数据");

        //收到消息后用来处理消息的回调对象
        DeliverCallback callback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                String msg = new String(message.getBody(), "UTF-8");
                String routingKey = message.getEnvelope().getRoutingKey();
                System.out.println("收到: "+routingKey+" - "+msg);
            }
        };

        //消费者取消时的回调对象
        CancelCallback cancel = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
            }
        };

        ch.basicConsume(queueName, true, callback, cancel);
    }
}

 

7.6、RPC模式

如果我们需要在远程电脑上运行一个方法,并且还要等待一个返回结果该怎么办?这和前面的例子不太一样, 这种模式我们通常称为远程过程调用,即RPC.

在本节中,我们将会学习使用RabbitMQ去搭建一个RPC系统:一个客户端和一个可以升级(扩展)的RPC服务器。为了模拟一个耗时任务,我们将创建一个返回斐波那契数列的虚拟的RPC服务。

客户端

在客户端定义一个RPCClient类,并定义一个call()方法,这个方法发送一个RPC请求,并等待接收响应结果

RPCClient client = new RPCClient();
String result = client.call("4");
System.out.println( "第四个斐波那契数是: " + result);

回调队列 Callback Queue

使用RabbitMQ去实现RPC很容易。一个客户端发送请求信息,并得到一个服务器端回复的响应信息。为了得到响应信息,我们需要在请求的时候发送一个“回调”队列地址。我们可以使用默认队列。下面是示例代码:

//定义回调队列,
//自动生成对列名,非持久,独占,自动删除
callbackQueueName = ch.queueDeclare().getQueue();

//用来设置回调队列的参数对象
BasicProperties props = new BasicProperties
 .Builder()
 .replyTo(callbackQueueName)
 .build();
//发送调用消息
ch.basicPublish("", "rpc_queue", props, message.getBytes());

消息属性 Message Properties
AMQP 0-9-1协议定义了消息的14个属性。大部分属性很少使用,下面是比较常用的4个:

deliveryMode:将消息标记为持久化(值为2)或非持久化(任何其他值)。

contentType:用于描述mime类型。例如,对于经常使用的JSON格式,将此属性设置为:application/json。

replyTo:通常用于指定回调队列。

correlationId:将RPC响应与请求关联起来非常有用。

关联id (correlationId):

在上面的代码中,我们会为每个RPC请求创建一个回调队列。 这是非常低效的,这里还有一个更好的方法:让我们为每个客户端创建一个回调队列。

这就提出了一个新的问题,在队列中得到一个响应时,我们不清楚这个响应所对应的是哪一条请求。这时候就需要使用关联id(correlationId)。我们将为每一条请求设置唯一的的id值。稍后,当我们在回调队列里收到一条消息的时候,我们将查看它的id属性,这样我们就可以匹配对应的请求和响应。如果我们发现了一个未知的id值,我们可以安全的丢弃这条消息,因为它不属于我们的请求。

RPC的工作方式是这样的:

  • 对于RPC请求,客户端发送一条带有两个属性的消息:replyTo,设置为仅为请求创建的匿名独占队列,和correlationId,设置为每个请求的惟一id值。
  • 请求被发送到rpc_queue队列。
  • RPC工作进程(即:服务器)在队列上等待请求。当一个请求出现时,它执行任务,并使用replyTo字段中的队列将结果发回客户机。
  • 客户机在回应消息队列上等待数据。当消息出现时,它检查correlationId属性。如果匹配请求中的值,则向程序返回该响应数据。

服务端代码

package rabbitmq.rpc;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.*;
import java.io.IOException;

/**
 * RabbitMQ  PRC模式服务端
 */
public class RPCServer {

    public static void main(String[] args) throws Exception {
        ConnectionFactory f = new ConnectionFactory();
        f.setHost("192.168.71.128");
        f.setPort(5672);
        f.setUsername("admin");
        f.setPassword("admin");

        Connection c = f.newConnection();
        Channel ch = c.createChannel();
        /*
         * 定义队列 rpc_queue, 将从它接收请求信息
         *
         * 参数:
         * 1. queue, 对列名
         * 2. durable, 持久化
         * 3. exclusive, 排他
         * 4. autoDelete, 自动删除
         * 5. arguments, 其他参数属性
         */
        ch.queueDeclare("rpc_queue",false,false,false,null);
        ch.queuePurge("rpc_queue");//清除队列中的内容

        ch.basicQos(1);//一次只接收一条消息


        //收到请求消息后的回调对象
        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                //处理收到的数据(要求第几个斐波那契数)
                String msg = new String(message.getBody(), "UTF-8");
                int n = Integer.parseInt(msg);
                //求出第n个斐波那契数
                int r = fbnq(n);
                String response = String.valueOf(r);

                //设置发回响应的id, 与请求id一致, 这样客户端可以把该响应与它的请求进行对应
                BasicProperties replyProps = new BasicProperties.Builder()
                        .correlationId(message.getProperties().getCorrelationId())
                        .build();
                /*
                 * 发送响应消息
                 * 1. 默认交换机
                 * 2. 由客户端指定的,用来传递响应消息的队列名
                 * 3. 参数(关联id)
                 * 4. 发回的响应消息
                 */
                ch.basicPublish("",message.getProperties().getReplyTo(), replyProps, response.getBytes("UTF-8"));
                //发送确认消息
                ch.basicAck(message.getEnvelope().getDeliveryTag(), false);
            }
        };

        //
        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
            }
        };

        //消费者开始接收消息, 等待从 rpc_queue接收请求消息, 不自动确认
        ch.basicConsume("rpc_queue", false, deliverCallback, cancelCallback);
    }

    protected static int fbnq(int n) {
        if(n == 1 || n == 2) return 1;

        return fbnq(n-1)+fbnq(n-2);
    }

}

 

RPC模式客户端

package rabbitmq.rpc;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.Scanner;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
/**
 * RabbitMQ  RPC模式客户端
 */
public class RPCClient {

    Connection con;
    Channel ch;

    public RPCClient() throws Exception {
        ConnectionFactory f = new ConnectionFactory();
        f.setHost("192.168.71.128");
        f.setUsername("admin");
        f.setPassword("admin");
        con = f.newConnection();
        ch = con.createChannel();
    }

    public String call(String msg) throws Exception {
        //自动生成对列名,非持久,独占,自动删除
        String replyQueueName = ch.queueDeclare().getQueue();
        //生成关联id
        String corrId = UUID.randomUUID().toString();

        //设置两个参数:
        //1. 请求和响应的关联id
        //2. 传递响应数据的queue
        BasicProperties props = new BasicProperties.Builder()
                .correlationId(corrId)
                .replyTo(replyQueueName)
                .build();
        //向 rpc_queue 队列发送请求数据, 请求第n个斐波那契数
        ch.basicPublish("", "rpc_queue", props, msg.getBytes("UTF-8"));

        //用来保存结果的阻塞集合,取数据时,没有数据会暂停等待
        BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);

        //接收响应数据的回调对象
        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                //如果响应消息的关联id,与请求的关联id相同,我们来处理这个响应数据
                if (message.getProperties().getCorrelationId().contentEquals(corrId)) {
                    //把收到的响应数据,放入阻塞集合
                    response.offer(new String(message.getBody(), "UTF-8"));
                }
            }
        };

        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
            }
        };

        //开始从队列接收响应数据
        ch.basicConsume(replyQueueName, true, deliverCallback, cancelCallback);
        //返回保存在集合中的响应数据
        return response.take();
    }

    public static void main(String[] args) throws Exception {
        RPCClient client = new RPCClient();
        while (true) {
            System.out.print("求第几个斐波那契数:");
            int n = new Scanner(System.in).nextInt();
            String r = client.call(""+n);
            System.out.println(r);
        }
    }
}

 

 

 

 

赞(0) 打赏
未经允许不得转载:所有资源由个人搜集整理自互联网,仅用于学习交流使用,请勿用于其他商业用途。如有侵权,请联系2448941602@qq.com删除科技云分享 » 『Java』RabbitMQ消息队列相关配置及基本使用(六种模式)
分享到: 更多 (0)

评论 抢沙发

评论前必须登录!

 

更多资源,尽在微信公众号【科技云分享】

关注公众号加入Q群

我叫科技云,感谢您的支持~

支付宝扫一扫打赏

微信扫一扫打赏