RabbitMQ In JAVA 介绍及使用

介绍:

  RabbitMQ是开源的消息中间件,它是轻量级的,支持多种消息传递协议,可以部署在分布式和联合配置中,以满足高级别、高可用性需求。并且可在许多操作系统和云环境上运行,并为大多数流行语言提供了广泛的开发工具。(这里只介绍JAVA下的RabbitMQ的使用,感兴趣的可以查看官方文档:http://www.rabbitmq.com/getstarted.html);

安装:

  参考:http://www.cnblogs.com/lfalex0831/p/8951955.html(windows安装)

应用场景:

  1、异步处理,主要为了较少请求的响应时间和解耦。即将比较耗时又不需要同步的操作放入消息队列中进行传递请求,只要保证消息格式(可以理解为接头的暗号)不变,这样消息的发送方和接收方互不干扰交互,即为解耦;

  2、广播,顾名思义,广播的好处就是一次发送,大家共享,大大的减少了冗余的操作,也降低了新增消费者的成本;

  3、流量削峰,比如秒杀活动,因为流量过大,导致应用挂掉,为了避免这个问题,会在应用前端加入消息队列。    作用:      1.可以控制进入后台的服务,超过阀值的订单直接丢弃;     2.可以缓解瞬时的高流量压垮应用;

  ps:秒杀系统优化思路可以从将请求尽量拦截在系统上游充分利用缓存;

  (如果还有别的应用场景,请大家多多指教。。。)

使用场景(本文使用的RabbitMQ的版本为5.20版本):

This tutorial assumes RabbitMQ is installed and running on localhost on standard port (5672). In case you use a different host, port or credentials, connections settings would require adjusting.

注意:根据官方文档说明,RabbitMQ的默认访问端口为5672,而管理端口为15672,希望不要搞混了(我刚接触时就没注意,果断乱了。。-_-||)。

  基本概念:

  • Broker:简单来说就是消息队列服务器实体
  • Exchange:消息交换机,它指定消息按什么规则,路由到哪个队列
  • Queue:消息队列载体,每个消息都会被投入到一个或多个队列
  • Binding:绑定,它的作用就是把exchange和queue按照路由规则绑定起来
  • Routing Key:路由关键字,exchange根据这个关键字进行消息投递
  • vhost:虚拟主机,一个broker里可以开设多个vhost,用作不同用户的权限分离
  • producer:消息生产者,就是投递消息的程序
  • consumer:消息消费者,就是接受消息的程序
  • channel:消息通道,在客户端的每个连接里,可建立多个channel,每个channel代表一个会话任务

  1、简单使用(HelloWorld)

  在下图中,P是我们的生产者,C是我们的消费者。中间的框是一个队列——RabbitMQ代表消费者保存的消息缓冲区。

  创建RabbitMQ的工厂类:

 1 import com.rabbitmq.client.Connection;
 2 import com.rabbitmq.client.ConnectionFactory;
 3 import java.io.IOException;
 4 import java.util.concurrent.TimeoutException;
 5 
 6 public class ConnectionUtil {
 7 
 8     private static final String RABBIT_HOST = "localhost";
 9 
10     private static final String RABBIT_USERNAME = "guest";
11 
12     private static final String RABBIT_PASSWORD = "guest";
13 
14     private static Connection connection = null;
15 
16     public static Connection getConnection() {
17         if(connection == null) {
18             ConnectionFactory connectionFactory = new ConnectionFactory();
19             connectionFactory.setHost(RABBIT_HOST);
20             connectionFactory.setUsername(RABBIT_USERNAME);
21             connectionFactory.setPassword(RABBIT_PASSWORD);
22             try {
23                 connection = connectionFactory.newConnection();
24             } catch (IOException e) {
25                 e.printStackTrace();
26             } catch (TimeoutException e) {
27                 e.printStackTrace();
28             }
29         }
30         return connection;
31     }
32 
33 }

创建生产者Producer:

 1 import com.cn.ConnectionUtil;
 2 import com.rabbitmq.client.Channel;
 3 import com.rabbitmq.client.Connection;
 4 import java.io.IOException;
 5 import java.util.concurrent.TimeoutException;
 6 
 7 public class Producer {
 8 
 9     private static final String QUEUE_NAME="test_queue";
10 
11     public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
12         //获取连接
13         Connection connection = ConnectionUtil.getConnection();
14         System.out.println(connection);
15         //创建通道
16         Channel channel = connection.createChannel(1);
17         /*
18          * 声明(创建)队列
19          * 参数1:队列名称
20          * 参数2:为true时server重启队列不会消失
21          * 参数3:队列是否是独占的,如果为true只能被一个connection使用,其他连接建立时会抛出异常
22          * 参数4:队列不再使用时是否自动删除(没有连接,并且没有未处理的消息)
23          * 参数5:建立队列时的其他参数
24          */
25         channel.queueDeclare(QUEUE_NAME,false,false,false,null);
26         String message = "Hello World!";
27         for (int i = 0; i < 20; i++) {
28             message = message + i;
29             channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
30             Thread.sleep(1000);
31         }
32         System.out.println("生产者 send :"+message);
33         channel.close();
34         connection.close();
35     }
36 
37 }

  创建消费者Consumer:

 1 import com.cn.ConnectionUtil;
 2 import com.rabbitmq.client.AMQP.BasicProperties;
 3 import com.rabbitmq.client.Channel;
 4 import com.rabbitmq.client.Connection;
 5 import com.rabbitmq.client.DefaultConsumer;
 6 import com.rabbitmq.client.Envelope;
 7 import java.io.IOException;
 8 
 9 public class Consumer {
10 
11     private static final String QUEUE_NAME = "test_queue";
12 
13     public static void main(String[] args) throws IOException {
14         Connection connection = ConnectionUtil.getConnection();
15         Channel channel = connection.createChannel(1);
16         channel.queueDeclare(QUEUE_NAME,false,false,false,null);
17         StringBuffer message = new StringBuffer();
18         //自4.0+ 版本后无法再使用QueueingConsumer,而官方推荐使用DefaultConsumer
19         com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
20             @Override
21             public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
22                 throws IOException {
23                 try {
24                     Thread.sleep(2000);
25                 } catch (InterruptedException e) {
26                     e.printStackTrace();
27                 }
28                 super.handleDelivery(consumerTag, envelope, properties, body);
29                 message.append(new String(body,"UTF-8"));
30                 System.out.println(new String(body,"UTF-8"));
31             }
32         };
33         //监听队列,当b为true时,为自动提交(只要消息从队列中获取,无论消费者获取到消息后是否成功消息,都认为是消息已经成功消费),
34         // 当b为false时,为手动提交(消费者从队列中获取消息后,服务器会将该消息标记为不可用状态,等待消费者的反馈,
35         // 如果消费者一直没有反馈,那么该消息将一直处于不可用状态。
36         //如果选用自动确认,在消费者拿走消息执行过程中出现宕机时,消息可能就会丢失!!)
37         //使用channel.basicAck(envelope.getDeliveryTag(),false);进行消息确认
38         channel.basicConsume(QUEUE_NAME,true,consumer);
39         System.out.println(message.toString());
40     }
41 }

  测试结果,Consumer收到Producer的消息。

  2、Work Queue

  上一个例子是一对一发送接收形式,而工作队列为一对多发送接收形式。工作队列(即任务队列)背后的主要思想是避免立即执行资源密集型任务,并且必须等待它完成。相反,我们把任务安排在以后做。我们将任务封装为消息并将其发送到队列。在后台运行的工作进程会弹出任务并最终执行任务。当你运行许多Consumer时,任务将在他们之间共享,如下图:

  由于工厂类已经创建,直接使用即可。

  创建生产者Producer:

package com.cn.work;

import com.cn.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

/**
 * @program: rabbit-learn
 * @description: 生产者
 * @create: 2018-04-26 16:18
 **/
public class Producer {

    private final static String QUEUE_NAME = "test_queue_work";

    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        for (int i = 0; i < 50; i++) {
            String message = "" + i;
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            Thread.sleep(100 * i);
        }
        channel.close();
        connection.close();
    }

}

  创建消费者1,2:

import com.cn.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;

public class Consumer1 {

    private final static String QUEUE_NAME = "test_queue_work";

    public static void main(String[] args) throws IOException {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.basicQos(1);//能者多劳模式
        //声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        //自4.0+ 版本后无法再使用QueueingConsumer,而官方推荐使用DefaultConsumer
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
                String message = new String(body,"UTF-8");
                System.out.println(message);
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                try {
                    doWork(message);
                }finally {
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            }
        };
        //监听队列,当b为true时,为自动提交(只要消息从队列中获取,无论消费者获取到消息后是否成功消息,都认为是消息已经成功消费),
        // 当b为false时,为手动提交(消费者从队列中获取消息后,服务器会将该消息标记为不可用状态,等待消费者的反馈,
        // 如果消费者一直没有反馈,那么该消息将一直处于不可用状态。
        //如果选用自动确认,在消费者拿走消息执行过程中出现宕机时,消息可能就会丢失!!)
        //使用channel.basicAck(envelope.getDeliveryTag(),false);进行消息确认
        channel.basicConsume(QUEUE_NAME,false,consumer);
    }

    private static void doWork(String task) {
        for (char ch : task.toCharArray()) {
            if (ch == '.') {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException _ignored) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

}
 1 import com.cn.ConnectionUtil;
 2 import com.rabbitmq.client.AMQP.BasicProperties;
 3 import com.rabbitmq.client.Channel;
 4 import com.rabbitmq.client.Connection;
 5 import com.rabbitmq.client.Consumer;
 6 import com.rabbitmq.client.DefaultConsumer;
 7 import com.rabbitmq.client.Envelope;
 8 import java.io.IOException;
 9 
10 public class Consumer2 {
11 
12     private final static String QUEUE_NAME = "test_queue_work";
13 
14     public static void main(String[] args) throws IOException {
15         Connection connection = ConnectionUtil.getConnection();
16         Channel channel = connection.createChannel();
17         channel.basicQos(1);//能者多劳模式
18         //声明队列
19         channel.queueDeclare(QUEUE_NAME,false,false,false,null);
20         //自4.0+ 版本后无法再使用QueueingConsumer,而官方推荐使用DefaultConsumer
21         Consumer consumer = new DefaultConsumer(channel) {
22             @Override
23             public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
24                 throws IOException {
25                 super.handleDelivery(consumerTag, envelope, properties, body);
26                 String message = new String(body,"UTF-8");
27                 System.out.println(message);
28                 try {
29                     Thread.sleep(1000);
30                 } catch (InterruptedException e) {
31                     e.printStackTrace();
32                 }
33                 try {
34                     doWork(message);
35                 }finally {
36                     channel.basicAck(envelope.getDeliveryTag(), false);
37                 }
38             }
39         };
40         //监听队列,当b为true时,为自动提交(只要消息从队列中获取,无论消费者获取到消息后是否成功消息,都认为是消息已经成功消费),
41         // 当b为false时,为手动提交(消费者从队列中获取消息后,服务器会将该消息标记为不可用状态,等待消费者的反馈,
42         // 如果消费者一直没有反馈,那么该消息将一直处于不可用状态。
43         //如果选用自动确认,在消费者拿走消息执行过程中出现宕机时,消息可能就会丢失!!)
44         //使用channel.basicAck(envelope.getDeliveryTag(),false);进行消息确认
45         channel.basicConsume(QUEUE_NAME,false,consumer);
46     }
47 
48     /** 
49      * @Description: 业务代码
50      * @Param:  
51      * @return:  
52      * @Author: 535504 
53      * @Date: 2018/4/26 
54      */ 
55     private static void doWork(String task) {
56         for (char ch : task.toCharArray()) {
57             if (ch == '.') {
58                 try {
59                     Thread.sleep(1000);
60                 } catch (InterruptedException _ignored) {
61                     Thread.currentThread().interrupt();
62                 }
63             }
64         }
65     }
66 
67 }

  测试结果,当消费者中的channel.basicQos(1);这行代码的注释打开时,执行会发现,休眠时间短的消费者执行的任务多,而注释后,再次执行会发现消费者1和消费者2获取到的消息内容是不同的,同一个消息只能被一个消费者获取,消费者1和消费者2获取到的消息的数量是相同的,一个是奇数一个是偶数。

  3、Publish/Subscribe(严格来说下面介绍的路由和通配符模式也是发布订阅)

  在发布订阅模式中,消息需要发送到MQ的交换机exchange上,exchange根据配置的路由方式发到相应的Queue上,Queue又将消息发送给consumer,消息从queue到consumer, 消息队列的使用过程大概如下:   1.客户端连接到消息队列服务器,打开一个channel。   2.客户端声明一个exchange,并设置相关属性。   3.客户端声明一个queue,并设置相关属性。   4.客户端在exchange和queue之间建立好绑定关系。   5.客户端投递消息到exchange。

  创建生产者Producer:

 1 import com.cn.ConnectionUtil;
 2 import com.rabbitmq.client.Channel;
 3 import com.rabbitmq.client.Connection;
 4 import java.io.IOException;
 5 import java.util.concurrent.TimeoutException;
 6 
 7 /**
 8  * @program: rabbit-learn
 9  * @description: 生产者,订阅模式
10  * @author: 
11  * @create: 
12  * 消息发送到没有队列绑定的交换机时,消息将丢失,因为,交换机没有存储消息的能力,消息只能存在在队列中。
13  **/
14 public class Producer {
15 
16     //交换机名称
17     private static final String EXCHANGE_NAME = "test_exchange_fanout";
18 
19     public static void main(String[] args) throws IOException, TimeoutException {
20         Connection connection = ConnectionUtil.getConnection();
21         Channel channel = connection.createChannel();
22         /*
23             声明exchange交换机
24             参数1:交换机名称
25             参数2:交换机类型
26             参数3:交换机持久性,如果为true则服务器重启时不会丢失
27             参数4:交换机在不被使用时是否删除
28             参数5:交换机的其他属性
29          */
30         channel.exchangeDeclare(EXCHANGE_NAME,"fanout", true,true,null);
31 
32         String message = "订阅消息";
33         channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes());
34         System.out.println("生产者 send :"+message);
35         channel.close();
36         connection.close();
37     }
38 
39 }

  创建消费者Consumer1、2:

 1 import com.cn.ConnectionUtil;
 2 import com.rabbitmq.client.AMQP.BasicProperties;
 3 import com.rabbitmq.client.Channel;
 4 import com.rabbitmq.client.Connection;
 5 import com.rabbitmq.client.Consumer;
 6 import com.rabbitmq.client.DefaultConsumer;
 7 import com.rabbitmq.client.Envelope;
 8 import java.io.IOException;
 9 
10 public class Consumer1 {
11 
12     private static final String QUEUE_NAME = "test_queue_exchange_1";
13 
14     private static final String EXCHANGE_NAME = "test_exchange_fanout";
15 
16     public static void main(String[] args) throws IOException {
17         Connection connection = ConnectionUtil.getConnection();
18         Channel channel = connection.createChannel();
19 
20         //声明队列
21         channel.queueDeclare(QUEUE_NAME,false,false,false,null);
22 
23         /*
24             绑定队列到交换机(这个交换机名称一定要和生产者的交换机名相同)
25             参数1:队列名
26             参数2:交换机名
27             参数3:Routing key 路由键
28          */
29         channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"");
30 
31         //同一时刻服务器只会发一条数据给消费者
32         channel.basicQos(1);
33 
34         Consumer consumer = new DefaultConsumer(channel) {
35             @Override
36             public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
37                 throws IOException {
38                 super.handleDelivery(consumerTag, envelope, properties, body);
39                 String message = new String(body,"UTF-8");
40                 System.out.println("收到消息:"+message);
41                 channel.basicAck(envelope.getDeliveryTag(),false);
42             }
43         };
44         channel.basicConsume(QUEUE_NAME,false,consumer);
45     }
46 
47 }
 1 import com.cn.ConnectionUtil;
 2 import com.rabbitmq.client.AMQP.BasicProperties;
 3 import com.rabbitmq.client.Channel;
 4 import com.rabbitmq.client.Connection;
 5 import com.rabbitmq.client.Consumer;
 6 import com.rabbitmq.client.DefaultConsumer;
 7 import com.rabbitmq.client.Envelope;
 8 import java.io.IOException;
 9 
10 public class Consumer2 {
11 
12     private static final String QUEUE_NAME = "test_queue_exchange_2";
13 
14     private static final String EXCHANGE_NAME = "test_exchange_fanout";
15 
16     public static void main(String[] args) throws IOException {
17         Connection connection = ConnectionUtil.getConnection();
18         Channel channel = connection.createChannel();
19 
20         //声明队列
21         channel.queueDeclare(QUEUE_NAME,false,false,false,null);
22 
23         /*
24             绑定队列到交换机(这个交换机名称一定要和生产者的交换机名相同)
25             参数1:队列名
26             参数2:交换机名
27             参数3:Routing key 路由键
28          */
29         channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"");
30 
31         //同一时刻服务器只会发一条数据给消费者
32         channel.basicQos(1);
33 
34         Consumer consumer = new DefaultConsumer(channel) {
35             @Override
36             public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
37                 throws IOException {
38                 super.handleDelivery(consumerTag, envelope, properties, body);
39                 String message = new String(body,"UTF-8");
40                 System.out.println("收到消息:"+message);
41                 channel.basicAck(envelope.getDeliveryTag(),false);
42             }
43         };
44         channel.basicConsume(QUEUE_NAME,false,consumer);
45     }
46 
47 }

  4、Routing(路由)

  根据指定的路由键发送到对应的消息队列中,如下图,在这个设置中,我们可以看到与它绑定的两个队列的直接交换X。第一个队列绑定了绑定键橙色,第二个队列有两个绑定,一个绑定键为黑色,另一个为绿色。在这样的设置中,将发送到与路由键橙色的交换的消息将被路由到队列Q1。带有黑色或绿色的路由键的消息将会进入Q2。所有其他消息将被丢弃。

  创建生产者Producer:

 1 import com.cn.ConnectionUtil;
 2 import com.rabbitmq.client.Channel;
 3 import com.rabbitmq.client.Connection;
 4 
 5 public class Producer {
 6 
 7     private static final String EXCHANGE_NAME = "test_exchange_direct";
 8 
 9     public static void main(String[] argv) throws Exception {
10         // 获取到连接以及mq通道
11         Connection connection = ConnectionUtil.getConnection();
12         Channel channel = connection.createChannel();
13 
14         // 声明exchange,路由模式声明direct
15         channel.exchangeDeclare(EXCHANGE_NAME, "direct");
16 
17         // 消息内容
18         String message = "这是消息B";
19         channel.basicPublish(EXCHANGE_NAME, "B", null, message.getBytes());
20         String messageA = "这是消息A";
21         channel.basicPublish(EXCHANGE_NAME, "A", null, messageA.getBytes());
22         System.out.println(" [生产者] Sent '" + message + "'");
23 
24         channel.close();
25         connection.close();
26     }
27 
28 }

  创建消费者Consumer1、2:

 1 package com.cn.routing;
 2 
 3 import com.cn.ConnectionUtil;
 4 import com.rabbitmq.client.AMQP.BasicProperties;
 5 import com.rabbitmq.client.Channel;
 6 import com.rabbitmq.client.Connection;
 7 import com.rabbitmq.client.Consumer;
 8 import com.rabbitmq.client.DefaultConsumer;
 9 import com.rabbitmq.client.Envelope;
10 import java.io.IOException;
11 
12 /**
13  * @program: rabbit-learn
14  * @description: 消费者1
15  * @author: 535504
16  * @create: 2018-04-26 17:52
17  **/
18 public class Consumer1 {
19     private final static String QUEUE_NAME = "test_queue_direct_A";
20 
21     private final static String EXCHANGE_NAME = "test_exchange_direct";
22 
23     public static void main(String[] argv) throws Exception {
24 
25         // 获取到连接以及mq通道
26         Connection connection = ConnectionUtil.getConnection();
27         Channel channel = connection.createChannel();
28 
29         // 声明队列
30         //channel.queueDeclare(QUEUE_NAME, false, false, false, null);
31         channel.exchangeDeclare(EXCHANGE_NAME,"direct");
32         /*
33          * 绑定队列到交换机
34          * 参数1:队列的名称
35          * 参数2:交换机的名称
36          * 参数3:routingKey
37          */
38         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "A");
39 
40         // 同一时刻服务器只会发一条消息给消费者
41         channel.basicQos(1);
42 
43         // 定义队列的消费者
44         Consumer consumer = new DefaultConsumer(channel) {
45             @Override
46             public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
47                 throws IOException {
48                 super.handleDelivery(consumerTag, envelope, properties, body);
49                 System.out.println(new String(body,"UTF-8"));
50             }
51         };
52         channel.basicConsume(QUEUE_NAME,true,consumer);
53     }
54 }
 1 import com.cn.ConnectionUtil;
 2 import com.rabbitmq.client.AMQP.BasicProperties;
 3 import com.rabbitmq.client.Channel;
 4 import com.rabbitmq.client.Connection;
 5 import com.rabbitmq.client.Consumer;
 6 import com.rabbitmq.client.DefaultConsumer;
 7 import com.rabbitmq.client.Envelope;
 8 import java.io.IOException;
 9 
10 public class Consumer2 {
11     private final static String QUEUE_NAME = "test_queue_direct_B";
12 
13     private final static String EXCHANGE_NAME = "test_exchange_direct";
14 
15     public static void main(String[] argv) throws Exception {
16 
17         // 获取到连接以及mq通道
18         Connection connection = ConnectionUtil.getConnection();
19         Channel channel = connection.createChannel();
20 
21         // 声明队列
22         //channel.queueDeclare(QUEUE_NAME, false, false, false, null);
23         channel.exchangeDeclare(EXCHANGE_NAME,"direct");
24         /*
25          * 绑定队列到交换机
26          * 参数1:队列的名称
27          * 参数2:交换机的名称
28          * 参数3:routingKey
29          */
30         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "B");
31 
32         // 同一时刻服务器只会发一条消息给消费者
33         channel.basicQos(1);
34 
35         // 定义队列的消费者
36         Consumer consumer = new DefaultConsumer(channel) {
37             @Override
38             public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
39                 throws IOException {
40                 super.handleDelivery(consumerTag, envelope, properties, body);
41                 System.out.println(new String(body,"UTF-8"));
42             }
43         };
44         channel.basicConsume(QUEUE_NAME,true,consumer);
45     }
46 }

  5、Topics(主题通配符)

  可以理解为Routing的通配符模式,如下图:

  “#”:表示匹配一个或多个词;(lazy.a.b.c)

  “*”:表示匹配一个词;(a.orange.b)

  创建生产者Producer:

 1 import com.cn.ConnectionUtil;
 2 import com.rabbitmq.client.Channel;
 3 import com.rabbitmq.client.Connection;
 4 
 5 import java.io.IOException;
 6 import java.util.concurrent.TimeoutException;
 7 
 8 public class Producer {
 9 
10     private static final String EXCHANGE_NAME = "test_exchange_topic";
11 
12     public static void main(String[] args) throws IOException, TimeoutException {
13         Connection connection = ConnectionUtil.getConnection();
14         Channel channel = connection.createChannel();
15         //声明交换机
16         channel.exchangeDeclare(EXCHANGE_NAME,"topic");
17         String message = "匹配insert";
18         channel.basicPublish(EXCHANGE_NAME,"order.update",false,false,null,message.getBytes());
19 
20         channel.close();
21         connection.close();
22     }
23 }

  创建消费者Consumer1、2:

 1 import com.cn.ConnectionUtil;
 2 import com.rabbitmq.client.AMQP.BasicProperties;
 3 import com.rabbitmq.client.Channel;
 4 import com.rabbitmq.client.Connection;
 5 import com.rabbitmq.client.Consumer;
 6 import com.rabbitmq.client.DefaultConsumer;
 7 import com.rabbitmq.client.Envelope;
 8 
 9 import java.io.IOException;
10 
11 public class Consumer1 {
12 
13     private static final String EXCHANGE_NAME = "test_exchange_topic";
14 
15     private  static final String QUEUE_NAME = "test_queue_topic_1";
16 
17     public static void main(String[] args) throws IOException {
18         Connection connection = ConnectionUtil.getConnection();
19         Channel channel = connection.createChannel();
20         //channel.queueDeclare(QUEUE_NAME,false,false,false,null);
21         channel.exchangeDeclare(EXCHANGE_NAME,"topic");
22         //order.#
23         channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"order.*");
24 
25         channel.basicQos(1);
26 
27         Consumer consumer = new DefaultConsumer(channel) {
28             @Override
29             public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
30                 throws IOException {
31                 super.handleDelivery(consumerTag, envelope, properties, body);
32                 System.out.println(new String(body,"UTF-8"));
33             }
34         };
35         channel.basicConsume(QUEUE_NAME,true,consumer);
36 
37     }
38 }
 1 import com.cn.ConnectionUtil;
 2 import com.rabbitmq.client.AMQP.BasicProperties;
 3 import com.rabbitmq.client.Channel;
 4 import com.rabbitmq.client.Connection;
 5 import com.rabbitmq.client.Consumer;
 6 import com.rabbitmq.client.DefaultConsumer;
 7 import com.rabbitmq.client.Envelope;
 8 
 9 import java.io.IOException;
10 
11 public class Consumer2 {
12 
13     private static final String EXCHANGE_NAME = "test_exchange_topic";
14 
15     private  static final String QUEUE_NAME = "test_queue_topic_2";
16 
17     public static void main(String[] args) throws IOException {
18         Connection connection = ConnectionUtil.getConnection();
19         Channel channel = connection.createChannel();
20         channel.queueDeclare(QUEUE_NAME,false,false,false,null);
21 
22         channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"order.insert");
23 
24         channel.basicQos(1);
25 
26         Consumer consumer = new DefaultConsumer(channel) {
27             @Override
28             public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
29                 super.handleDelivery(consumerTag, envelope, properties, body);
30                 System.out.println("接收消息:" + new String(body, "UTF-8"));
31             }
32         };
33         channel.basicConsume(QUEUE_NAME,true,consumer);
34 
35     }
36 }

  6、RPC(远程调用)

  如果我们需要在远程计算机上运行一个函数并等待结果,这种模式通常称为远程过程调用或RPC;

  创建RPC服务:

 1 import com.cn.ConnectionUtil;
 2 import com.rabbitmq.client.AMQP;
 3 import com.rabbitmq.client.Channel;
 4 import com.rabbitmq.client.Connection;
 5 import com.rabbitmq.client.Consumer;
 6 import com.rabbitmq.client.DefaultConsumer;
 7 import com.rabbitmq.client.Envelope;
 8 
 9 import java.io.IOException;
10 import java.util.concurrent.TimeoutException;
11 
12 public class RPCServer {
13 
14     private static final String RPC_QUEUE_NAME = "rpc_queue";
15 
16     public static void main(String[] args) throws IOException, TimeoutException {
17         Connection connection = ConnectionUtil.getConnection();
18         final Channel channel = connection.createChannel();
19         channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
20         channel.basicQos(1);
21 
22         Consumer consumer = new DefaultConsumer(channel) {
23             @Override
24             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
25                 super.handleDelivery(consumerTag, envelope, properties, body);
26                 AMQP.BasicProperties properties1 = new AMQP.BasicProperties.Builder().correlationId(properties.getCorrelationId()).build();
27                 String mes = new String(body, "UTF-8");
28                 int num = Integer.valueOf(mes);
29                 System.out.println("接收数据:" + num);
30                 num = fib(num);
31                 channel.basicPublish("", properties.getReplyTo(), properties1, String.valueOf(num).getBytes());
32                 channel.basicAck(envelope.getDeliveryTag(), false);
33             }
34         };
35         channel.basicConsume(RPC_QUEUE_NAME, false, consumer);
36         while (true) {
37             synchronized (consumer) {
38                 try {
39                     consumer.wait();
40                 } catch (InterruptedException e) {
41                     e.printStackTrace();
42                 }
43             }
44         }
45     }
46 
47     /*
48         获取斐波那契数列的第n个值得大小
49      */
50     private static int fib(int n) {
51         System.out.println(n);
52         if (n == 0)
53             return 0;
54         if (n == 1)
55             return 1;
56         return fib(n - 1) + fib(n - 2);
57     }
58 }

  创建RPC客户端:

 1 import com.cn.ConnectionUtil;
 2 import com.rabbitmq.client.AMQP;
 3 import com.rabbitmq.client.Channel;
 4 import com.rabbitmq.client.Connection;
 5 import com.rabbitmq.client.DefaultConsumer;
 6 import com.rabbitmq.client.Envelope;
 7 
 8 import java.io.IOException;
 9 import java.util.UUID;
10 import java.util.concurrent.ArrayBlockingQueue;
11 import java.util.concurrent.BlockingQueue;
12 import java.util.concurrent.TimeoutException;
13 
14 public class RPCClient {
15 
16     private Connection connection;
17     private Channel channel;
18     private String requestQueueName = "rpc_queue";
19     private String replyQueueName;
20 
21     public RPCClient() throws IOException, TimeoutException {
22         connection = ConnectionUtil.getConnection();
23         channel = connection.createChannel();
24 
25         replyQueueName = channel.queueDeclare().getQueue();
26     }
27 
28     public String call(String message) throws IOException, InterruptedException {
29         final String corrId = UUID.randomUUID().toString();
30 
31         AMQP.BasicProperties props = new AMQP.BasicProperties
32                 .Builder()
33                 .correlationId(corrId)
34                 .replyTo(replyQueueName)
35                 .build();
36 
37         channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));
38 
39         final BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);
40 
41         channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) {
42             @Override
43             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
44                 if (properties.getCorrelationId().equals(corrId)) {
45                     response.offer(new String(body, "UTF-8"));
46                 }
47             }
48         });
49 
50         //close();
51         return response.take();
52     }
53 
54     public void close() throws IOException {
55         connection.close();
56     }
57 
58 }

  创建RPC测试类:

 1 import java.io.IOException;
 2 import java.util.concurrent.TimeoutException;
 3 
 4 public class RPCTest {
 5 
 6     public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
 7         RPCClient rpcClient = new RPCClient();
 8         System.out.println(rpcClient.call("2"));
 9     }
10 }

完整示例https://gitee.com/lfalex/rabbitmq-learn

参考官方文档:http://www.rabbitmq.com/getstarted.html

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

扫码关注云+社区

领取腾讯云代金券