消息队列-RabbitMQ

安装rabbitMQ

双击运行otp_win64_24.3.4.6.exe,点击下一步下一步即可,可以更改安装位置,需要注意路径不要有中文或空格

他的作用类似JDK,因为rabbitMQ是erlang语言开发的,所以需要环境支持

配置环境变量,在path中配置: 自己安装erlang的路径\bin
image-20230224202431336
image-20230224202301804

双击运行rabbitmq-server-3.10.1.exe,下一步即可,安装好后进入以下目录
image-20230224202733157
在该目录下打开dos窗口,输入以下运行命令

1
rabbitmq-plugins enable rabbitmq_management

启动结束后,访问:http://localhost:15672
image-20230224203737805
用户名和密码都是:guest
image-20230224203828259

rabbitMQ的基本概念

image-20230224210652678

Exchange

接受生产者发送的消息,并根据binding规则将消息路由给服务器中的队列。

ExchangeType决定了Exchange路由消息的行为。在RabbitMQ中,ExchangeType常用的类型有三种:direct、fanout和topic。

Message Queue

消息队列。我们发送给RabbitMQ的消息最后都会到达各种queue,并且存储在其中(如果路由找不到对应的queue则数据会丢失),等待消费者来取。

Binding Key

它表示的是Exchange和Message Queue是通过binding key进行联系的,这个关系是固定。

Routing Key

生产者在将消息发送给Exchange的时候,一般会指定一个routing key,来指定这个消息的路由规则。这个routing key需要与ExchangeType及binding key联合使用才能生效,我们的生产者只需要通过指定routing key来决定消息流向哪里

RabbitMQ六种工作模式

创建一个RabbitMQ的项目,用来学习rabbitmq,需要添加依赖

1
2
3
4
5
6
7
8
9
10
11
<!--添加RabbitMQ依赖-->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.12.0</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.6.4</version>
</dependency>

简单模式

只有一个消费者
image-20230224212545820

生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
package cn.tedu.rabbitmq.simple;

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

//模拟简单模式的生产者
public class Producer {
public static void main(String[] args)throws Exception{
//1.配置RabbitMQ的连接信息
ConnectionFactory cf = new ConnectionFactory();
cf.setHost("127.0.0.1");
cf.setPort(5672);//15672是UI端口,5672是客户端端口(可以省略)
cf.setUsername("guest");
cf.setPassword("guest");
//2.创建连接以及channel
Connection nc = cf.newConnection();
Channel cc = nc.createChannel();
//3.创建队列
/*
第一个参数:队列名称
第二个参数:是否是一个持久队列
第三个参数:是否是一个独占队列
第四个参数:是否自动删除
第五个参数:其他参数属性的设置
*/
cc.queueDeclare("helloworld",false,false,false,null);
//4.准备数据
String msg = "hello world" + System.currentTimeMillis();
//5.发送数据
/*
第一个参数:先忽略
第二个参数:先写成队列名称
第三个参数:其他属性设置
第四个参数:消息数据,需要转成byte[]类型
*/
cc.basicPublish("","helloworld",null,msg.getBytes());
nc.close();
}
}

消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
package cn.tedu.rabbitmq.simple;

import com.rabbitmq.client.*;

import java.io.IOException;

//简单模式的消费者
public class Consumer {
public static void main(String[] args) throws Exception{
//1.配置连接信息
ConnectionFactory cf = new ConnectionFactory();
cf.setHost("127.0.0.1");
cf.setUsername("guest");
cf.setPassword("guest");
//2.创建连接以及channel
Connection nc = cf.newConnection();
Channel cc = nc.createChannel();
//3.创建队列
//如果该队列已经在服务器中存在,那么会忽略创建该队列的命令
cc.queueDeclare("helloworld",false,false,false,null);
//4.消费消息
//处理数据的回调函数
DeliverCallback deliverCallback = new DeliverCallback() {
@Override
public void handle(String s, Delivery delivery) throws IOException {
byte[] body = delivery.getBody();
String msg = new String(body);
System.out.println("收到了消息:"+msg);
System.out.println("===========================");
}
};
//取消接收数据的回调函数
CancelCallback cancelCallback = new CancelCallback() {
@Override
public void handle(String s) throws IOException {

}
};
cc.basicConsume("helloworld",true,deliverCallback,cancelCallback);
}
}

工作队列模式

多个消费者,从同一个队列中接受消息

负载均衡,消息会轮询发送给所有消费者

合理的分发消息

  1. 手动ack

    消息回执

    向服务器发送一个通知,告诉服务器一条消息已经处理完毕

    服务器可以通过ack,知道消费者是空闲还是繁忙

  2. qos=1

​ 每次抓取的消息数量

​ 消息处理完毕之前,不会抓取新消息

​ 手动ack模式下才有效

消息持久化:在服务器端,持久保存消息,避免因为服务器宕机而造成消息丢失

  1. 队列持久化: cc.queueDeclare(“队列名”,true,……)
  2. 消息持久化: cc.basocPublish(“”,”队列名”,MessageProperties.PERSISTENT_TEXT_PLAIN,消息)

生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
package cn.tedu.rabbitmq.work;

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

import java.util.Scanner;

//模拟工作队列模式的生产者
public class Producer {
public static void main(String[] args)throws Exception{
//1.配置RabbitMQ的连接信息
ConnectionFactory cf = new ConnectionFactory();
cf.setHost("127.0.0.1");
cf.setPort(5672);//15672是UI端口,5672是客户端端口(可以省略)
cf.setUsername("guest");
cf.setPassword("guest");
//2.创建连接以及channel
Connection nc = cf.newConnection();
Channel cc = nc.createChannel();
//3.创建队列
/*
第一个参数:队列名称
第二个参数:是否是一个持久队列
第三个参数:是否是一个独占队列
第四个参数:是否自动删除
第五个参数:其他参数属性的设置
*/
cc.queueDeclare("work_queue",false,false,false,null);
//4.准备数据,发送数据
while(true){
System.out.print("输入消息:");
String msg = new Scanner(System.in).nextLine();
if (msg.equals("quit"))
break;
cc.basicPublish("","work_queue",null,msg.getBytes());
}
nc.close();
}
}

消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
package cn.tedu.rabbitmq.work;

import com.rabbitmq.client.*;

import java.io.IOException;

//工作队列模式的消费者
public class Consumer {
public static void main(String[] args) throws Exception{
//1.配置连接信息
ConnectionFactory cf = new ConnectionFactory();
cf.setHost("127.0.0.1");
cf.setUsername("guest");
cf.setPassword("guest");
//2.创建连接以及channel
Connection nc = cf.newConnection();
Channel cc = nc.createChannel();
//3.创建队列
//如果该队列已经在服务器中存在,那么会忽略创建该队列的命令
cc.queueDeclare("work_queue",false,false,false,null);
//4.消费消息
//处理数据的回调函数
DeliverCallback deliverCallback = new DeliverCallback() {
@Override
public void handle(String s, Delivery delivery) throws IOException {
byte[] body = delivery.getBody();
String msg = new String(body);
System.out.println("收到了消息:"+msg);
//处理消息的时候,遇到一个‘.’,暂停1s,模拟实际处理消息的阻塞场景
for (int i = 0; i < msg.length(); i++) {
//charAt(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 {

}
};
cc.basicConsume("work_queue",true,deliverCallback,cancelCallback);
}
}

合理分发

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
package cn.tedu.rabbitmq.work;

import com.rabbitmq.client.*;

import java.io.IOException;

//工作队列模式的消费者
public class Consumer {
public static void main(String[] args) throws Exception{
//1.配置连接信息
ConnectionFactory cf = new ConnectionFactory();
cf.setHost("127.0.0.1");
cf.setUsername("guest");
cf.setPassword("guest");
//2.创建连接以及channel
Connection nc = cf.newConnection();
Channel cc = nc.createChannel();
//3.创建队列
//如果该队列已经在服务器中存在,那么会忽略创建该队列的命令
cc.queueDeclare("work_queue",false,false,false,null);
//4.消费消息
//处理数据的回调函数
DeliverCallback deliverCallback = new DeliverCallback() {
@Override
public void handle(String s, Delivery delivery) throws IOException {
byte[] body = delivery.getBody();
String msg = new String(body);
System.out.println("收到了消息:"+msg);
//处理消息的时候,遇到一个‘.’,暂停1s,模拟实际处理消息的阻塞场景
for (int i = 0; i < msg.length(); i++) {
//charAt(i)
if(msg.charAt(i)=='.'){
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
System.out.println("消息处理完毕");
//发送消息确认(消息回执)
//第一个参数:消息的标签,需要从消息中获取
//第二个参数:是否确认多条信息,false,只确认一条消息
cc.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
}
};
//取消接收数据的回调函数
CancelCallback cancelCallback = new CancelCallback() {
@Override
public void handle(String s) throws IOException {

}
};
//第二个参数:ack的设置,默认是autoAck,false是手动ack
//设置消费者每次只获取一条消息
cc.basicQos(1);
cc.basicConsume("work_queue",false,deliverCallback,cancelCallback);
}
}

持久化

如果消息保存在服务器端的内存中,是不安全的,如果服务器宕机,信息会丢失

1
2
停止rabbitmq服务:rabbitmq-service stop或者rabbitmqctl stop
启动rabbitmq服务:rabbitmq-service start

消息数据持久化、消息队列持久化

但是需要注意持久化队列时,不能直接修改队列属性,因为队列一定创建属性就固定了,不可以进行修改

可以选择新建一个别名队列或者删除该队列重新创建

队列持久化:cc.queueDeclare(“task_queue”,true,false,false,null);

数据持久化:cc.basicPublish(“”,”task_queue”, MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes());

发布订阅模式

把消息群发给所有消费者,同一条消息所有消费者都可以收到

fanout类型的交换机

生产者:定义交换机,向交换机发送数据

消费者:定义交换机,定义自已独占的非持久的自动删除随机队列,绑定,正常从随机队列中接受数据

生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
package cn.tedu.rabbitmq.publishsubscribe;

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

import java.util.Scanner;

//模拟发布订阅模式的生产者
public class Producer {
public static void main(String[] args) throws Exception{
//1.配置RabbitMQ的连接信息
ConnectionFactory cf = new ConnectionFactory();
cf.setHost("127.0.0.1");
cf.setPort(5672);//15672是UI端口,5672是客户端端口(可以省略)
cf.setUsername("guest");
cf.setPassword("guest");
//2.创建连接以及channel
Connection nc = cf.newConnection();
Channel cc = nc.createChannel();
//3.定义交换机
//如果服务器中没有,就创建,有就直接使用
//第一个参数:交换机名称
//第二个参数:交换机类型
cc.exchangeDeclare("ps_exchange","fanout");
//4.向交换机发送数据,交换机只是接受发送数据,并不保存数据
//如果没有消费者接受,数据会丢失
while (true){
System.out.print("输入消息:");
String msg = new Scanner(System.in).nextLine();
if (msg.equals("quit"))
break;
cc.basicPublish("ps_exchange","",null,msg.getBytes());
}
nc.close();
}
}

消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
package cn.tedu.rabbitmq.publishsubscribe;

import com.rabbitmq.client.*;

import java.io.IOException;

//发布订阅模式的消费者
public class Consumer {
public static void main(String[] args) throws Exception{
//1.配置RabbitMQ的连接信息
ConnectionFactory cf = new ConnectionFactory();
cf.setHost("127.0.0.1");
cf.setPort(5672);//15672是UI端口,5672是客户端端口(可以省略)
cf.setUsername("guest");
cf.setPassword("guest");
//2.创建连接以及channel
Connection nc = cf.newConnection();
Channel cc = nc.createChannel();
//3.定义交换机
//如果服务器中没有,就创建,有就直接使用
//第一个参数:交换机名称
//第二个参数:交换机类型
cc.exchangeDeclare("ps_exchange","fanout");
//4.定义队列--随机队列名,否持久,独占、自动删除的
String queue = cc.queueDeclare().getQueue();//无参构造就可以满足我们的需求
//5.绑定交换机和队列
/*
第一个参数:队列名称
第二个参数:交换机名称
第三个参数:队列和交换机绑定的关系
*/
cc.queueBind(queue,"ps_exchange","");

//6.处理接收到的消息
//处理数据
DeliverCallback deliverCallback = new DeliverCallback() {
@Override
public void handle(String s, Delivery delivery) throws IOException {
byte[] body = delivery.getBody();
String msg = new String(body);
System.out.println("收到:"+msg);
System.out.println("消息处理完毕");
}
};
//取消数据时的回调函数
CancelCallback cancelCallback = new CancelCallback() {
@Override
public void handle(String s) throws IOException {

}
};
cc.basicConsume(queue,true,deliverCallback,cancelCallback);
}
}

路由模式

通过关键字匹配,来决定把消息发送到哪个队列

生产者:定义direct类型的交换机,向交换机发送数据并携带路由键

消费者:定义交换机,定义队列,用绑定键来绑定交换机和队列,正常接收消息

生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
package cn.tedu.rabbitmq.route;

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

import java.util.Scanner;

//模拟路由模式的生产者
public class Producer {
public static void main(String[] args) throws Exception{
//1.配置RabbitMQ的连接信息
ConnectionFactory cf = new ConnectionFactory();
cf.setHost("127.0.0.1");
cf.setPort(5672);//15672是UI端口,5672是客户端端口(可以省略)
cf.setUsername("guest");
cf.setPassword("guest");
//2.创建连接以及channel
Connection nc = cf.newConnection();
Channel cc = nc.createChannel();
//3.定义direct类型的交换机
cc.exchangeDeclare("route_exchange", BuiltinExchangeType.DIRECT);
//4.发送消息,携带路由键
while(true){
Scanner scanner = new Scanner(System.in);
System.out.print("输入消息:");
String msg = scanner.nextLine();
System.out.print("路由键:");
String routingKey = scanner.nextLine();
if(msg.equals("quit"))
break;
cc.basicPublish("route_exchange",routingKey,null,msg.getBytes());
System.out.println("消息发送成功");
}
nc.close();
}
}

消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
package cn.tedu.rabbitmq.route;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.Scanner;

//模拟路由模式的消费者
public class Consumer {
public static void main(String[] args) throws Exception{
//1.配置RabbitMQ的连接信息
ConnectionFactory cf = new ConnectionFactory();
cf.setHost("127.0.0.1");
cf.setPort(5672);//15672是UI端口,5672是客户端端口(可以省略)
cf.setUsername("guest");
cf.setPassword("guest");
//2.创建连接以及channel
Connection nc = cf.newConnection();
Channel cc = nc.createChannel();
//3.定义direct类型的交换机
cc.exchangeDeclare("route_exchange", BuiltinExchangeType.DIRECT);
//4.定义队列
String queue = cc.queueDeclare().getQueue();
//5.绑定交换机和队列(重复绑定多次)
System.out.print("输入绑定键,用空格隔开:");
String bindingKeys = new Scanner(System.in).nextLine();
String[] keys = bindingKeys.split("\\s+");
for (String bindingKey : keys) {
cc.queueBind(queue,"route_exchange",bindingKey);
}
//处理数据
DeliverCallback deliverCallback = new DeliverCallback() {
@Override
public void handle(String s, Delivery delivery) throws IOException {
String msg = new String(delivery.getBody());
String routingKey = delivery.getEnvelope().getRoutingKey();
System.out.println(msg+"-"+routingKey);
System.out.println("====================");
}
};
CancelCallback cancelCallback = new CancelCallback() {
@Override
public void handle(String s) throws IOException {

}
};
cc.basicConsume(queue,true,deliverCallback,cancelCallback);
}
}

主题模式

和路由模式相同,具有特殊的关键字规则

topic类型的交换机实现这种特殊路由规则

aaa.bbb.ccc.ddd

*.ccc.ddd.eee

.ddd

“*”可以通配单个单词

“#”可以通配零个或多个单词

生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
package cn.tedu.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;

//模拟主题模式的生产者
public class Producer {
public static void main(String[] args) throws Exception{
//1.配置RabbitMQ的连接信息
ConnectionFactory cf = new ConnectionFactory();
cf.setHost("127.0.0.1");
cf.setPort(5672);//15672是UI端口,5672是客户端端口(可以省略)
cf.setUsername("guest");
cf.setPassword("guest");
//2.创建连接以及channel
Connection nc = cf.newConnection();
Channel cc = nc.createChannel();
//3.定义direct类型的交换机
cc.exchangeDeclare("topic_exchange", BuiltinExchangeType.TOPIC);
//4.发送消息,携带路由键
while(true){
Scanner scanner = new Scanner(System.in);
System.out.print("输入消息:");
String msg = scanner.nextLine();
System.out.print("路由键:");
String routingKey = scanner.nextLine();
if(msg.equals("quit"))
break;
cc.basicPublish("topic_exchange",routingKey,null,msg.getBytes());
System.out.println("消息发送成功");
}
nc.close();
}
}

消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
package cn.tedu.rabbitmq.topic;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.Scanner;

//模拟主题模式的消费者
public class Consumer {
public static void main(String[] args) throws Exception{
//1.配置RabbitMQ的连接信息
ConnectionFactory cf = new ConnectionFactory();
cf.setHost("127.0.0.1");
cf.setPort(5672);//15672是UI端口,5672是客户端端口(可以省略)
cf.setUsername("guest");
cf.setPassword("guest");
//2.创建连接以及channel
Connection nc = cf.newConnection();
Channel cc = nc.createChannel();
//3.定义direct类型的交换机
cc.exchangeDeclare("topic_exchange", BuiltinExchangeType.TOPIC);
//4.定义队列
String queue = cc.queueDeclare().getQueue();
//5.绑定交换机和队列(重复绑定多次)
System.out.print("输入绑定键,用空格隔开:");
String bindingKeys = new Scanner(System.in).nextLine();
String[] keys = bindingKeys.split("\\s+");
for (String bindingKey : keys) {
cc.queueBind(queue,"topic_exchange",bindingKey);
}
//处理数据
DeliverCallback deliverCallback = new DeliverCallback() {
@Override
public void handle(String s, Delivery delivery) throws IOException {
String msg = new String(delivery.getBody());
String routingKey = delivery.getEnvelope().getRoutingKey();
System.out.println(msg+"-"+routingKey);
System.out.println("====================");
}
};
CancelCallback cancelCallback = new CancelCallback() {
@Override
public void handle(String s) throws IOException {
}
};
cc.basicConsume(queue,true,deliverCallback,cancelCallback);
}
}

RPC模式

实现原理

  1. 两个队列

    调用队列

    返回队列:每个客户端,都需要有自己的返回队列

  2. 返回队列的队列名

    在调用消息数据中,携带返回队列名

    根据返回队列名,向正确的返回队列来发送计算结果

  3. 关联ID

    用来匹配计算结果和调用

    如果连续发送多次调用,获得计算结果时,需要直到这个节骨欧式按一次调用的结果

    客户端发送调用时,携带一个关联ID

    服务器端返回结果时,也携带这个关联ID

客户端:多线程异步处理结果

  1. 主线程

    发送调用信息

    需要计算结果时,要取结果

  2. 次线程:等待接受结果

  3. 线程之间传递数据,可以使用BlockingQueue

    集合工具

    这个集合中,添加了线程的等待和通知

    如果没有数据,取数据时会暂停等待

    有多个子类:比如ArrayBlockingQueue

服务器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
package cn.tedu.rabbitmq.rpc;

import com.rabbitmq.client.*;
import com.rabbitmq.client.AMQP.BasicProperties;

import java.io.IOException;

public class RPCServer {
public static void main(String[] args) throws Exception{
ConnectionFactory cf = new ConnectionFactory();
cf.setHost("127.0.0.1");
cf.setPort(5672);//15672是UI端口,5672是客户端端口(可以省略)
cf.setUsername("guest");
cf.setPassword("guest");
Connection nc = cf.newConnection();
Channel cc = nc.createChannel();
//1.接受客户端发送的调用信息(正整数n)
//2.执行算法求第n个斐波那契数
//3.向客户端发送计算结果
//定义调用队列
cc.queueDeclare("rpc_queue",false,false,false,null);
//从调用队列中取调用信息
DeliverCallback deliverCallback = new DeliverCallback() {
@Override
public void handle(String s, Delivery delivery) throws IOException {
//从delivery取出正整数n
String msg = new String(delivery.getBody());
String replyTo = delivery.getProperties().getReplyTo();//返回队列名
String correlationId = delivery.getProperties().getCorrelationId();//关联id
long fbnqs = fbnqs(Integer.parseInt(msg));
BasicProperties basicProperties = new BasicProperties.Builder().correlationId(correlationId).build();
cc.basicPublish("",replyTo,basicProperties,(""+fbnqs).getBytes());
}
};
CancelCallback cancelCallback = new CancelCallback() {
@Override
public void handle(String s) throws IOException {

}
};
cc.basicConsume("rpc_queue",true,deliverCallback,cancelCallback);

}
//服务:求第n个斐波那契数
//1 1 2 3 5 8 13 21 34 55 ......
//递归效率低,可以用来模拟服务器端的耗时运算
static long fbnqs(int n){
if(n==1 || n==2)
return 1;
return fbnqs(n-1)+fbnqs(n-2);
}
}

客户端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
package cn.tedu.rabbitmq.rpc;

import com.rabbitmq.client.*;
import com.rabbitmq.client.AMQP.BasicProperties;

import java.io.IOException;
import java.util.Scanner;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;


public class RPCClient {
static BlockingQueue<Long> q = new ArrayBlockingQueue<Long>(10);
public static void main(String[] args) throws Exception{
System.out.print("输入求第几个斐波那契数:");
int n = new Scanner(System.in).nextInt();
long fbnqs = fbnqs(n);
System.out.println("第"+n+"个的斐波那契数是:"+fbnqs);
}
//异步调用服务器,从服务器中获取结果
private static long fbnqs(int n) throws Exception{
ConnectionFactory cf = new ConnectionFactory();
cf.setHost("127.0.0.1");
cf.setPort(5672);//15672是UI端口,5672是客户端端口(可以省略)
cf.setUsername("guest");
cf.setPassword("guest");
Connection nc = cf.newConnection();
Channel cc = nc.createChannel();
//定义调用队列
cc.queueDeclare("rpc_queue",false,false,false,null);
//返回队列
String replyTo = cc.queueDeclare().getQueue();
//关联id
String cid = UUID.randomUUID().toString();
BasicProperties basicProperties = new BasicProperties.Builder()
.replyTo(replyTo)
.correlationId(cid)
.build();
cc.basicPublish("","rpc_queue",basicProperties,(""+n).getBytes());
//模拟执行其他运算,不等待计算结果
System.out.println("调用消息已经发送");
System.out.println("模拟执行其他运算,不立即等待计算结果");
//获取计算结果
DeliverCallback deliverCallback = new DeliverCallback() {
@Override
public void handle(String s, Delivery delivery) throws IOException {
//处理数据之前,先对比关联id
if(cid.equals(delivery.getProperties().getCorrelationId())){
String msg = new String(delivery.getBody());
long fbnqs = Integer.parseInt(msg);
q.offer(fbnqs);
cc.getConnection().close();
}
}
};
CancelCallback cancelCallback = new CancelCallback() {
@Override
public void handle(String s) throws IOException {

}
};
cc.basicConsume(replyTo,true,deliverCallback,cancelCallback);
return q.take();
}
}