消息队列-RabbitMQ

消息队列-RabbitMQ#

安装rabbitMQ#

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

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

配置环境变量，在path中配置: 自己安装erlang的路径\bin

双击运行rabbitmq-server-3.10.1.exe，下一步即可，安装好后进入以下目录

在该目录下打开dos窗口，输入以下运行命令

1
rabbitmq-plugins enable rabbitmq_management

启动结束后，访问：http://localhost:15672

用户名和密码都是：guest

rabbitMQ的基本概念#

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

简单模式#

只有一个消费者

生产者#

1
package cn.tedu.rabbitmq.simple;
2

3
import com.rabbitmq.client.Channel;
4
import com.rabbitmq.client.Connection;
5
import com.rabbitmq.client.ConnectionFactory;
6

7
//模拟简单模式的生产者
8
public class Producer {
9
    public static void main(String[] args)throws Exception{
10
        //1.配置RabbitMQ的连接信息
11
        ConnectionFactory cf = new ConnectionFactory();
12
        cf.setHost("127.0.0.1");
13
        cf.setPort(5672);//15672是UI端口，5672是客户端端口(可以省略)
14
        cf.setUsername("guest");
15
        cf.setPassword("guest");
16
        //2.创建连接以及channel
17
        Connection nc = cf.newConnection();
18
        Channel cc = nc.createChannel();
19
        //3.创建队列
20
        /*
21
          第一个参数：队列名称
22
          第二个参数：是否是一个持久队列
23
          第三个参数：是否是一个独占队列
24
          第四个参数：是否自动删除
25
          第五个参数：其他参数属性的设置
26
         */
27
        cc.queueDeclare("helloworld",false,false,false,null);
28
        //4.准备数据
29
        String msg = "hello world" + System.currentTimeMillis();
30
        //5.发送数据
31
        /*
32
        第一个参数：先忽略
33
        第二个参数：先写成队列名称
34
        第三个参数：其他属性设置
35
        第四个参数：消息数据，需要转成byte[]类型
36
         */
37
        cc.basicPublish("","helloworld",null,msg.getBytes());
38
        nc.close();
39
    }
40
}

消费者#

1
package cn.tedu.rabbitmq.simple;
2

3
import com.rabbitmq.client.*;
4

5
import java.io.IOException;
6

7
//简单模式的消费者
8
public class Consumer {
9
    public static void main(String[] args) throws Exception{
10
        //1.配置连接信息
11
        ConnectionFactory cf = new ConnectionFactory();
12
        cf.setHost("127.0.0.1");
13
        cf.setUsername("guest");
14
        cf.setPassword("guest");
15
        //2.创建连接以及channel
16
        Connection nc = cf.newConnection();
17
        Channel cc = nc.createChannel();
18
        //3.创建队列
19
        //如果该队列已经在服务器中存在，那么会忽略创建该队列的命令
20
        cc.queueDeclare("helloworld",false,false,false,null);
21
        //4.消费消息
22
        //处理数据的回调函数
23
        DeliverCallback deliverCallback = new DeliverCallback() {
24
            @Override
25
            public void handle(String s, Delivery delivery) throws IOException {
26
                byte[] body = delivery.getBody();
27
                String msg = new String(body);
28
                System.out.println("收到了消息："+msg);
29
                System.out.println("===========================");
30
            }
31
        };
32
        //取消接收数据的回调函数
33
        CancelCallback cancelCallback = new CancelCallback() {
34
            @Override
35
            public void handle(String s) throws IOException {
36

37
            }
38
        };
39
        cc.basicConsume("helloworld",true,deliverCallback,cancelCallback);
40
    }
41
}

工作队列模式#

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

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

合理的分发消息

手动ack

消息回执

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

服务器可以通过ack，知道消费者是空闲还是繁忙
qos=1

每次抓取的消息数量

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

手动ack模式下才有效

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

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

生产者#

1
package cn.tedu.rabbitmq.work;
2

3
import com.rabbitmq.client.Channel;
4
import com.rabbitmq.client.Connection;
5
import com.rabbitmq.client.ConnectionFactory;
6

7
import java.util.Scanner;
8

9
//模拟工作队列模式的生产者
10
public class Producer {
11
    public static void main(String[] args)throws Exception{
12
        //1.配置RabbitMQ的连接信息
13
        ConnectionFactory cf = new ConnectionFactory();
14
        cf.setHost("127.0.0.1");
15
        cf.setPort(5672);//15672是UI端口，5672是客户端端口(可以省略)
16
        cf.setUsername("guest");
17
        cf.setPassword("guest");
18
        //2.创建连接以及channel
19
        Connection nc = cf.newConnection();
20
        Channel cc = nc.createChannel();
21
        //3.创建队列
22
        /*
23
          第一个参数：队列名称
24
          第二个参数：是否是一个持久队列
25
          第三个参数：是否是一个独占队列
26
          第四个参数：是否自动删除
27
          第五个参数：其他参数属性的设置
28
         */
29
        cc.queueDeclare("work_queue",false,false,false,null);
30
        //4.准备数据，发送数据
31
        while(true){
32
            System.out.print("输入消息：");
33
            String msg = new Scanner(System.in).nextLine();
34
            if (msg.equals("quit"))
35
                break;
36
            cc.basicPublish("","work_queue",null,msg.getBytes());
37
        }
38
        nc.close();
39
    }
40
}

消费者#

1
package cn.tedu.rabbitmq.work;
2

3
import com.rabbitmq.client.*;
4

5
import java.io.IOException;
6

7
//工作队列模式的消费者
8
public class Consumer {
9
    public static void main(String[] args) throws Exception{
10
        //1.配置连接信息
11
        ConnectionFactory cf = new ConnectionFactory();
12
        cf.setHost("127.0.0.1");
13
        cf.setUsername("guest");
14
        cf.setPassword("guest");
15
        //2.创建连接以及channel
16
        Connection nc = cf.newConnection();
17
        Channel cc = nc.createChannel();
18
        //3.创建队列
19
        //如果该队列已经在服务器中存在，那么会忽略创建该队列的命令
20
        cc.queueDeclare("work_queue",false,false,false,null);
21
        //4.消费消息
22
        //处理数据的回调函数
23
        DeliverCallback deliverCallback = new DeliverCallback() {
24
            @Override
25
            public void handle(String s, Delivery delivery) throws IOException {
26
                byte[] body = delivery.getBody();
27
                String msg = new String(body);
28
                System.out.println("收到了消息："+msg);
29
                //处理消息的时候，遇到一个‘.’，暂停1s，模拟实际处理消息的阻塞场景
30
                for (int i = 0; i < msg.length(); i++) {
31
                    //charAt(i)
32
                    if(msg.charAt(i)=='.'){
33
                        try {
34
                            Thread.sleep(1000);
35
                        } catch (InterruptedException e) {
36
                            e.printStackTrace();
37
                        }
38
                    }
39
                }
40
                System.out.println("消息处理完毕");
41
            }
42
        };
43
        //取消接收数据的回调函数
44
        CancelCallback cancelCallback = new CancelCallback() {
45
            @Override
46
            public void handle(String s) throws IOException {
47

48
            }
49
        };
50
        cc.basicConsume("work_queue",true,deliverCallback,cancelCallback);
51
    }
52
}

合理分发#

1
package cn.tedu.rabbitmq.work;
2

3
import com.rabbitmq.client.*;
4

5
import java.io.IOException;
6

7
//工作队列模式的消费者
8
public class Consumer {
9
    public static void main(String[] args) throws Exception{
10
        //1.配置连接信息
11
        ConnectionFactory cf = new ConnectionFactory();
12
        cf.setHost("127.0.0.1");
13
        cf.setUsername("guest");
14
        cf.setPassword("guest");
15
        //2.创建连接以及channel
16
        Connection nc = cf.newConnection();
17
        Channel cc = nc.createChannel();
18
        //3.创建队列
19
        //如果该队列已经在服务器中存在，那么会忽略创建该队列的命令
20
        cc.queueDeclare("work_queue",false,false,false,null);
21
        //4.消费消息
22
        //处理数据的回调函数
23
        DeliverCallback deliverCallback = new DeliverCallback() {
24
            @Override
25
            public void handle(String s, Delivery delivery) throws IOException {
26
                byte[] body = delivery.getBody();
27
                String msg = new String(body);
28
                System.out.println("收到了消息："+msg);
29
                //处理消息的时候，遇到一个‘.’，暂停1s，模拟实际处理消息的阻塞场景
30
                for (int i = 0; i < msg.length(); i++) {
31
                    //charAt(i)
32
                    if(msg.charAt(i)=='.'){
33
                        try {
34
                            Thread.sleep(1000);
35
                        } catch (InterruptedException e) {
36
                            e.printStackTrace();
37
                        }
38
                    }
39
                }
40
                System.out.println("消息处理完毕");
41
                //发送消息确认(消息回执)
42
                //第一个参数：消息的标签，需要从消息中获取
43
                //第二个参数：是否确认多条信息，false，只确认一条消息
44
                cc.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
45
            }
46
        };
47
        //取消接收数据的回调函数
48
        CancelCallback cancelCallback = new CancelCallback() {
49
            @Override
50
            public void handle(String s) throws IOException {
51

52
            }
53
        };
54
        //第二个参数：ack的设置，默认是autoAck，false是手动ack
55
        //设置消费者每次只获取一条消息
56
        cc.basicQos(1);
57
        cc.basicConsume("work_queue",false,deliverCallback,cancelCallback);
58
    }
59
}

持久化#

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

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

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

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

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

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

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

发布订阅模式#

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

fanout类型的交换机

生产者：定义交换机，向交换机发送数据

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

生产者#

1
package cn.tedu.rabbitmq.publishsubscribe;
2

3
import com.rabbitmq.client.Channel;
4
import com.rabbitmq.client.Connection;
5
import com.rabbitmq.client.ConnectionFactory;
6

7
import java.util.Scanner;
8

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

消费者#

1
package cn.tedu.rabbitmq.publishsubscribe;
2

3
import com.rabbitmq.client.*;
4

5
import java.io.IOException;
6

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

34
        //6.处理接收到的消息
35
        //处理数据
36
        DeliverCallback deliverCallback = new DeliverCallback() {
37
            @Override
38
            public void handle(String s, Delivery delivery) throws IOException {
39
                byte[] body = delivery.getBody();
40
                String msg = new String(body);
41
                System.out.println("收到："+msg);
42
                System.out.println("消息处理完毕");
43
            }
44
        };
45
        //取消数据时的回调函数
46
        CancelCallback cancelCallback = new CancelCallback() {
47
            @Override
48
            public void handle(String s) throws IOException {
49

50
            }
51
        };
52
        cc.basicConsume(queue,true,deliverCallback,cancelCallback);
53
    }
54
}

路由模式#

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

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

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

生产者#

1
package cn.tedu.rabbitmq.route;
2

3
import com.rabbitmq.client.BuiltinExchangeType;
4
import com.rabbitmq.client.Channel;
5
import com.rabbitmq.client.Connection;
6
import com.rabbitmq.client.ConnectionFactory;
7

8
import java.util.Scanner;
9

10
//模拟路由模式的生产者
11
public class Producer {
12
    public static void main(String[] args) throws Exception{
13
        //1.配置RabbitMQ的连接信息
14
        ConnectionFactory cf = new ConnectionFactory();
15
        cf.setHost("127.0.0.1");
16
        cf.setPort(5672);//15672是UI端口，5672是客户端端口(可以省略)
17
        cf.setUsername("guest");
18
        cf.setPassword("guest");
19
        //2.创建连接以及channel
20
        Connection nc = cf.newConnection();
21
        Channel cc = nc.createChannel();
22
        //3.定义direct类型的交换机
23
        cc.exchangeDeclare("route_exchange", BuiltinExchangeType.DIRECT);
24
        //4.发送消息，携带路由键
25
        while(true){
26
            Scanner scanner = new Scanner(System.in);
27
            System.out.print("输入消息:");
28
            String msg = scanner.nextLine();
29
            System.out.print("路由键:");
30
            String routingKey = scanner.nextLine();
31
            if(msg.equals("quit"))
32
                break;
33
            cc.basicPublish("route_exchange",routingKey,null,msg.getBytes());
34
            System.out.println("消息发送成功");
35
        }
36
        nc.close();
37
    }
38
}

消费者#

1
package cn.tedu.rabbitmq.route;
2

3
import com.rabbitmq.client.*;
4

5
import java.io.IOException;
6
import java.util.Scanner;
7

8
//模拟路由模式的消费者
9
public class Consumer {
10
    public static void main(String[] args) throws Exception{
11
        //1.配置RabbitMQ的连接信息
12
        ConnectionFactory cf = new ConnectionFactory();
13
        cf.setHost("127.0.0.1");
14
        cf.setPort(5672);//15672是UI端口，5672是客户端端口(可以省略)
15
        cf.setUsername("guest");
16
        cf.setPassword("guest");
17
        //2.创建连接以及channel
18
        Connection nc = cf.newConnection();
19
        Channel cc = nc.createChannel();
20
        //3.定义direct类型的交换机
21
        cc.exchangeDeclare("route_exchange", BuiltinExchangeType.DIRECT);
22
        //4.定义队列
23
        String queue = cc.queueDeclare().getQueue();
24
        //5.绑定交换机和队列(重复绑定多次)
25
        System.out.print("输入绑定键，用空格隔开:");
26
        String bindingKeys = new Scanner(System.in).nextLine();
27
        String[] keys = bindingKeys.split("\\s+");
28
        for (String bindingKey : keys) {
29
            cc.queueBind(queue,"route_exchange",bindingKey);
30
        }
31
        //处理数据
32
        DeliverCallback deliverCallback = new DeliverCallback() {
33
            @Override
34
            public void handle(String s, Delivery delivery) throws IOException {
35
                String msg = new String(delivery.getBody());
36
                String routingKey = delivery.getEnvelope().getRoutingKey();
37
                System.out.println(msg+"-"+routingKey);
38
                System.out.println("====================");
39
            }
40
        };
41
        CancelCallback cancelCallback = new CancelCallback() {
42
            @Override
43
            public void handle(String s) throws IOException {
44

45
            }
46
        };
47
        cc.basicConsume(queue,true,deliverCallback,cancelCallback);
48
    }
49
}

主题模式#

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

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

aaa.bbb.ccc.ddd

*.ccc.ddd.eee

#.ddd

”*“可以通配单个单词

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

生产者#

1
package cn.tedu.rabbitmq.topic;
2

3
import com.rabbitmq.client.BuiltinExchangeType;
4
import com.rabbitmq.client.Channel;
5
import com.rabbitmq.client.Connection;
6
import com.rabbitmq.client.ConnectionFactory;
7

8
import java.util.Scanner;
9

10
//模拟主题模式的生产者
11
public class Producer {
12
    public static void main(String[] args) throws Exception{
13
        //1.配置RabbitMQ的连接信息
14
        ConnectionFactory cf = new ConnectionFactory();
15
        cf.setHost("127.0.0.1");
16
        cf.setPort(5672);//15672是UI端口，5672是客户端端口(可以省略)
17
        cf.setUsername("guest");
18
        cf.setPassword("guest");
19
        //2.创建连接以及channel
20
        Connection nc = cf.newConnection();
21
        Channel cc = nc.createChannel();
22
        //3.定义direct类型的交换机
23
        cc.exchangeDeclare("topic_exchange", BuiltinExchangeType.TOPIC);
24
        //4.发送消息，携带路由键
25
        while(true){
26
            Scanner scanner = new Scanner(System.in);
27
            System.out.print("输入消息:");
28
            String msg = scanner.nextLine();
29
            System.out.print("路由键:");
30
            String routingKey = scanner.nextLine();
31
            if(msg.equals("quit"))
32
                break;
33
            cc.basicPublish("topic_exchange",routingKey,null,msg.getBytes());
34
            System.out.println("消息发送成功");
35
        }
36
        nc.close();
37
    }
38
}

消费者#

1
package cn.tedu.rabbitmq.topic;
2

3
import com.rabbitmq.client.*;
4

5
import java.io.IOException;
6
import java.util.Scanner;
7

8
//模拟主题模式的消费者
9
public class Consumer {
10
    public static void main(String[] args) throws Exception{
11
        //1.配置RabbitMQ的连接信息
12
        ConnectionFactory cf = new ConnectionFactory();
13
        cf.setHost("127.0.0.1");
14
        cf.setPort(5672);//15672是UI端口，5672是客户端端口(可以省略)
15
        cf.setUsername("guest");
16
        cf.setPassword("guest");
17
        //2.创建连接以及channel
18
        Connection nc = cf.newConnection();
19
        Channel cc = nc.createChannel();
20
        //3.定义direct类型的交换机
21
        cc.exchangeDeclare("topic_exchange", BuiltinExchangeType.TOPIC);
22
        //4.定义队列
23
        String queue = cc.queueDeclare().getQueue();
24
        //5.绑定交换机和队列(重复绑定多次)
25
        System.out.print("输入绑定键，用空格隔开:");
26
        String bindingKeys = new Scanner(System.in).nextLine();
27
        String[] keys = bindingKeys.split("\\s+");
28
        for (String bindingKey : keys) {
29
            cc.queueBind(queue,"topic_exchange",bindingKey);
30
        }
31
        //处理数据
32
        DeliverCallback deliverCallback = new DeliverCallback() {
33
            @Override
34
            public void handle(String s, Delivery delivery) throws IOException {
35
                String msg = new String(delivery.getBody());
36
                String routingKey = delivery.getEnvelope().getRoutingKey();
37
                System.out.println(msg+"-"+routingKey);
38
                System.out.println("====================");
39
            }
40
        };
41
        CancelCallback cancelCallback = new CancelCallback() {
42
            @Override
43
            public void handle(String s) throws IOException {
44
            }
45
        };
46
        cc.basicConsume(queue,true,deliverCallback,cancelCallback);
47
    }
48
}

RPC模式#

实现原理

两个队列

调用队列

返回队列：每个客户端，都需要有自己的返回队列
返回队列的队列名

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

根据返回队列名，向正确的返回队列来发送计算结果
关联ID

用来匹配计算结果和调用

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

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

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

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

主线程

发送调用信息

需要计算结果时，要取结果
次线程：等待接受结果
线程之间传递数据，可以使用BlockingQueue

集合工具

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

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

有多个子类：比如ArrayBlockingQueue

服务器#

1
package cn.tedu.rabbitmq.rpc;
2

3
import com.rabbitmq.client.*;
4
import com.rabbitmq.client.AMQP.BasicProperties;
5

6
import java.io.IOException;
7

8
public class RPCServer {
9
    public static void main(String[] args) throws Exception{
10
        ConnectionFactory cf = new ConnectionFactory();
11
        cf.setHost("127.0.0.1");
12
        cf.setPort(5672);//15672是UI端口，5672是客户端端口(可以省略)
13
        cf.setUsername("guest");
14
        cf.setPassword("guest");
15
        Connection nc = cf.newConnection();
16
        Channel cc = nc.createChannel();
17
        //1.接受客户端发送的调用信息(正整数n)
18
        //2.执行算法求第n个斐波那契数
19
        //3.向客户端发送计算结果
20
        //定义调用队列
21
        cc.queueDeclare("rpc_queue",false,false,false,null);
22
        //从调用队列中取调用信息
23
        DeliverCallback deliverCallback = new DeliverCallback() {
24
            @Override
25
            public void handle(String s, Delivery delivery) throws IOException {
26
                //从delivery取出正整数n
27
                String msg = new String(delivery.getBody());
28
                String replyTo = delivery.getProperties().getReplyTo();//返回队列名
29
                String correlationId = delivery.getProperties().getCorrelationId();//关联id
30
                long fbnqs = fbnqs(Integer.parseInt(msg));
31
                BasicProperties basicProperties = new BasicProperties.Builder().correlationId(correlationId).build();
32
                cc.basicPublish("",replyTo,basicProperties,(""+fbnqs).getBytes());
33
            }
34
        };
35
        CancelCallback cancelCallback = new CancelCallback() {
36
            @Override
37
            public void handle(String s) throws IOException {
38

39
            }
40
        };
41
        cc.basicConsume("rpc_queue",true,deliverCallback,cancelCallback);
42

43
    }
44
    //服务：求第n个斐波那契数
45
    //1 1 2 3 5 8 13 21 34 55 ......
46
    //递归效率低，可以用来模拟服务器端的耗时运算
47
    static long fbnqs(int n){
48
        if(n==1 || n==2)
49
            return 1;
50
        return fbnqs(n-1)+fbnqs(n-2);
51
    }
52
}

客户端#

1
package cn.tedu.rabbitmq.rpc;
2

3
import com.rabbitmq.client.*;
4
import com.rabbitmq.client.AMQP.BasicProperties;
5

6
import java.io.IOException;
7
import java.util.Scanner;
8
import java.util.UUID;
9
import java.util.concurrent.ArrayBlockingQueue;
10
import java.util.concurrent.BlockingQueue;
11

12
public class RPCClient {
13
    static BlockingQueue<Long> q = new ArrayBlockingQueue<Long>(10);
14
    public static void main(String[] args) throws Exception{
15
        System.out.print("输入求第几个斐波那契数:");
16
        int n = new Scanner(System.in).nextInt();
17
        long fbnqs = fbnqs(n);
18
        System.out.println("第"+n+"个的斐波那契数是:"+fbnqs);
19
    }
20
    //异步调用服务器，从服务器中获取结果
21
    private static long fbnqs(int n) throws Exception{
22
        ConnectionFactory cf = new ConnectionFactory();
23
        cf.setHost("127.0.0.1");
24
        cf.setPort(5672);//15672是UI端口，5672是客户端端口(可以省略)
25
        cf.setUsername("guest");
26
        cf.setPassword("guest");
27
        Connection nc = cf.newConnection();
28
        Channel cc = nc.createChannel();
29
        //定义调用队列
30
        cc.queueDeclare("rpc_queue",false,false,false,null);
31
        //返回队列
32
        String replyTo = cc.queueDeclare().getQueue();
33
        //关联id
34
        String cid = UUID.randomUUID().toString();
35
        BasicProperties basicProperties = new BasicProperties.Builder()
36
                .replyTo(replyTo)
37
                .correlationId(cid)
38
                .build();
39
        cc.basicPublish("","rpc_queue",basicProperties,(""+n).getBytes());
40
        //模拟执行其他运算，不等待计算结果
41
        System.out.println("调用消息已经发送");
42
        System.out.println("模拟执行其他运算，不立即等待计算结果");
43
        //获取计算结果
44
        DeliverCallback deliverCallback = new DeliverCallback() {
45
            @Override
46
            public void handle(String s, Delivery delivery) throws IOException {
47
                //处理数据之前，先对比关联id
48
                if(cid.equals(delivery.getProperties().getCorrelationId())){
49
                    String msg = new String(delivery.getBody());
50
                    long fbnqs = Integer.parseInt(msg);
51
                    q.offer(fbnqs);
52
                    cc.getConnection().close();
53
                }
54
            }
55
        };
56
        CancelCallback cancelCallback = new CancelCallback() {
57
            @Override
58
            public void handle(String s) throws IOException {
59

60
            }
61
        };
62
        cc.basicConsume(replyTo,true,deliverCallback,cancelCallback);
63
        return q.take();
64
    }
65
}

消息队列-RabbitMQ