2. RabbitMQ消息模式种类

浅浅的花香味﹌ 2022-05-20 07:28 198阅读 0赞

pom依赖和log4j.properties

<dependencies>
    
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>4.5.0</version>
    </dependency>
    
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.25</version>
    </dependency>
    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-lang3</artifactId>
        <version>3.3.2</version>
    </dependency>
    
    
    <dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-rabbit</artifactId>
        <version>1.7.6.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-test</artifactId>
        <version>4.3.7.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>
    </dependencies>

log4j.properties

log4j.rootLogger=DEBUG,A1
    log4j.logger.com.taotao = DEBUG
    log4j.logger.org.mybatis = DEBUG
    log4j.appender.A1=org.apache.log4j.ConsoleAppender
    log4j.appender.A1.layout=org.apache.log4j.PatternLayout
    log4j.appender.A1.layout.ConversionPattern=%-d{yyyy-MM-dd HH:mm:ss,SSS} [%t] [%c]-
    [%p] %m%n

工具类：ConnectionUtil

public class ConnectionUtil {
        public static Connection getConnection() throws Exception {
            // 定义连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            // 设置服务地址
            factory.setHost("192.168.3.227");
            // 端口
            factory.setPort(5672);
            // 设置帐号信息，用户名，密码，vhost
            factory.setVirtualHost("/test");
            factory.setUsername("test");
            factory.setPassword("test");
            // 通过工厂获取连接
            Connection connection = factory.newConnection();
            return connection;
        }
    }

消息模式种类：  
[http://www.rabbitmq.com/getstarted.html][http_www.rabbitmq.com_getstarted.html]

每一个模式下面都有一个链接，里面也有详细的代码，可以参照

1.  简单模式

> [http://www.rabbitmq.com/tutorials/tutorial-one-java.html][http_www.rabbitmq.com_tutorials_tutorial-one-java.html]

生产者：

package com.rabbitmq.build;
    
    /** * Created by HCF&FCL on 2018/7/14. */
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    
    /** * 生产者 */
    public class Send { 
    
        private final static String QUEUE_NAME = "test_queue";
    
        public static void main(String[] argv) throws Exception {
            // 获取连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
    
            // 从连接中创建通道
            Channel channel = connection.createChannel();
    
            /** * 声明创建队列，如果不存在就创建，存在就不创建 * Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) throws IOException; * 参数说明： * String queue：队列名 * durable：是否持久化，队列声明默认是存放到内存中的，如果rabbitmq重启会丢失，如果重启之后还存在就要使用队列持久化，保存到Erlang自带的Mnesia数据库中,dang rabbitmq重启之后会读取该数据库 * exclusive：是否排外，作用： * 1. 当连接关闭时connection.close()该队列是否会自动删除； * 2. 该队列是否是私有的private ，如果不是排外的，可以使用两个消费者都访问同一个队列，没有任何问题，如果是排外的，就会对当前的队列加锁，其他通道channel是无法访问的，强制访问会报错： * com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=405, reply-text=RESOURCE_LOCKED - cannot obtain exclusive access to locked queue 'queue_name' in vhost '/', class-id=50, method-id=20) * 一般为true的话用于一个队列只能有一个消费者来消费的场景 * autoDelete：是否自动删除，当最后一个消费者断开连接之后对垒是否自动删除，可以通过 RabbitMQ Management 查看某个队列的消费者数量，当consumers= 0时会自动删除 * arguments：参数 */
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            String message = "hello world!";
            /** * basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body); * 参数说明： * exchange：交换机 * routingKey：发送到那个队列 * props：属性 * body：内容 */
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
    
            System.out.println("发送数据【" + message + "】");
    
            // 关闭通道和连接
            channel.close();
            connection.close();
    
        }
    }

消费者：

package com.rabbitmq.build;
    
    /** * Created by HCF&FCL on 2018/7/14. */
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    /** * 消费者 */
    public class Recv { 
    
        private final static String QUEUE_NAME = "test_queue";
    
        public static void main(String[] argv) throws Exception {
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
    
            // 监听队列，第二个参数自动确认消息模式
            channel.basicConsume(QUEUE_NAME, true, consumer);
    
            while (true) {
                // 消息获取完成后就会销毁， 如果没有消息会阻塞等待
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println("收到消息 【" + message + "】");
            }
        }
    }

运行send和recv

--------------------

> work模式：

[http://www.rabbitmq.com/tutorials/tutorial-two-java.html][http_www.rabbitmq.com_tutorials_tutorial-two-java.html]

提供者：

package com.rabbitmq.build;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    /** * Created by HCF&FCL on 2018/7/14. */
    
    /** * work模式 * 一条消息可以被多个消费者尝试接收，但是最终只能有一个消费者能获取 */
    public class WorkSend { 
        private final static String QUEUE_NAME = "test_queue_work";
    
        public static void main(String[] argv) throws Exception {
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 循环发送消息，但是每条消息时间间隔越来越长
            for (int i = 0; i < 100; i++) {
                String message = "hello world!" + i;
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                System.out.println("发送数据【" + message + "】");
                Thread.sleep(i * 10);
            }
            // 关闭通道和连接
            channel.close();
            connection.close();
    
        }
    }

消费者1：

package com.rabbitmq.build;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    /** * Created by HCF&FCL on 2018/7/14. */
    public class WorkRecv1 { 
    
        private final static String QUEUE_NAME = "test_queue_work";
    
        public static void main(String[] argv) throws Exception {
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 同一时刻服务器只会发一条消息给消费者，只有当前消费者将消息处理完之后才会获取到吓一跳消息
            // 注释掉可以获取多条消息，但是会逐条处理
            channel.basicQos(1);
    
            QueueingConsumer consumer = new QueueingConsumer(channel);
    
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            while (true) {
                // 消息获取完成后就会销毁， 如果没有消息会阻塞等待
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println("消费者1 收到消息 【" + message + "】");
                Thread.sleep(10);
    
                // 返回确认状态，手动确认模式，true代表拒绝消息
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

消费者2：

package com.rabbitmq.build;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    /** * Created by HCF&FCL on 2018/7/14. */
    public class WorkRecv2 { 
    
        private final static String QUEUE_NAME = "test_queue_work";
    
        public static void main(String[] argv) throws Exception {
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 同一时刻服务器只会发一条消息给消费者，只有当前消费者将消息处理完之后才会获取到吓一跳消息
            // 注释掉可以获取多条消息，但是会逐条处理
            channel.basicQos(1);
    
            QueueingConsumer consumer = new QueueingConsumer(channel);
    
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            while (true) {
                // 消息获取完成后就会销毁， 如果没有消息会阻塞等待
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println("消费者2 收到消息 【" + message + "】");
                Thread.sleep(1000);
    
                // 返回确认状态，手动确认模式，true代表拒绝消息
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

启动消费者1 2和提供者

发现：

channel.basicQos(1) 注释的情况下，接收到的消息数量是相等的

channel.basicQos(1)不注释的情况下，消费者1接受到的消息是明显多于消费者2的

--------------------

> 订阅模式：

[http://www.rabbitmq.com/tutorials/tutorial-three-java.html][http_www.rabbitmq.com_tutorials_tutorial-three-java.html]

提供者：

package com.rabbitmq.build;
    
    /** * Created by HCF&FCL on 2018/7/14. */
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    /** * 发布订阅模式生产者 */
    public class SubscribeSend { 
    
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] argv) throws Exception {
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
    
            String message = "Hello World";
            // 将消息发送到交换机，如果没有队列和该交换机绑定，则消息会丢失，因为交换机没有存储消息的能力
            // 所以启动的时候需要先启动消费者
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
            System.out.println("发送消息【" + message + "】");
    
            channel.close();
            connection.close();
        }
    }

消费者：

package com.rabbitmq.build;
    
    /** * Created by HCF&FCL on 2018/7/14. */
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    /** * 发布订阅模式消费者 */
    public class SubscribeRecv1 { 
    
        private final static String QUEUE_NAME = "test_queue_work1";
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] argv) throws Exception {
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
            channel.basicQos(1);
    
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听完成，手动发挥完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" 消费者1收到消息【" + message + "】");
                Thread.sleep(10);
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

package com.rabbitmq.build;
    
    /** * Created by HCF&FCL on 2018/7/14. */
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    /** * 发布订阅模式消费者 */
    public class SubscribeRecv2 { 
    
        private final static String QUEUE_NAME = "test_queue_work2";
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] argv) throws Exception {
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
            channel.basicQos(1);
    
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听完成，手动发挥完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" 消费者2收到消息【" + message + "】");
                Thread.sleep(10);
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

--------------------

> 路由模式

[http://www.rabbitmq.com/tutorials/tutorial-four-java.html][http_www.rabbitmq.com_tutorials_tutorial-four-java.html]

生产者将消息发送到了type为direct模式的交换机

消费者队列在将自己绑定到路由的时候会给自己绑定一个key

只有消费者发送对应key格式的消息时候队列才会收到消息

提供者：

package com.rabbitmq.build;
    
    /** * Created by HCF&FCL on 2018/7/14. */
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    /** * 路由模式提供者 */
    public class RoutingSend { 
    
        private final static String EXCHANGE_NAME = "test_exchange_direct";
    
        public static void main(String[] argv) throws Exception {
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明交换机，类型是direct
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
    
            String message = "Hello World!";
            // key设置为key2
            channel.basicPublish(EXCHANGE_NAME, "key2", null, message.getBytes()); //
    
            System.out.println(" 发送消息【" + message + "】");
            channel.close();
            connection.close();
        }
    }

消费者：

package com.rabbitmq.build;
    
    /** * Created by HCF&FCL on 2018/7/14. */
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    /** * 路由模式消费者1 */
    public class RoutingRecv1 { 
    
        private final static String QUEUE_NAME = "test_queue_work1";
        private final static String EXCHANGE_NAME = "test_exchange_direct";
    
        public static void main(String[] argv) throws Exception {
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定到交换机，设置对应的key，在绑定路由交换机的时候，该路由必须存在，所以必须先有提供者创建一个路由
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key");
            // 绑定多个key可以多次执行
    // channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key");
            channel.basicQos(1);
            QueueingConsumer consumer = new QueueingConsumer(channel);
            channel.basicConsume(QUEUE_NAME, false, consumer);
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" 消费者1收到消息【" + message + "】");
                Thread.sleep(10);
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

package com.rabbitmq.build;
    
    /** * Created by HCF&FCL on 2018/7/14. */
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    /** * 路由模式消费者1 */
    public class RoutingRecv2 { 
    
        private final static String QUEUE_NAME = "test_queue_work2";
        private final static String EXCHANGE_NAME = "test_exchange_direct";
    
        public static void main(String[] argv) throws Exception {
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定到交换机，设置对应的key，在绑定路由交换机的时候，该路由必须存在，所以必须先有提供者创建一个路由
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key2");
            // 绑定多个key可以多次执行
    // channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key");
            channel.basicQos(1);
            QueueingConsumer consumer = new QueueingConsumer(channel);
            channel.basicConsume(QUEUE_NAME, false, consumer);
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" 消费者2收到消息【" + message + "】");
                Thread.sleep(10);
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

--------------------

> 通配符模式：

[http://www.rabbitmq.com/tutorials/tutorial-five-java.html][http_www.rabbitmq.com_tutorials_tutorial-five-java.html]

就是将路由模式的key更改了，在消费者中的key改为“key.\*”或者”key.\#”

\*代表匹配一个词

# 代表匹配一个或者多个词 #

key.\# 能够匹配到 key.123.234.345.456

key.\* 只能匹配到 key.123

[http_www.rabbitmq.com_getstarted.html]: http://www.rabbitmq.com/getstarted.html
[http_www.rabbitmq.com_tutorials_tutorial-one-java.html]: http://www.rabbitmq.com/tutorials/tutorial-one-java.html
[http_www.rabbitmq.com_tutorials_tutorial-two-java.html]: http://www.rabbitmq.com/tutorials/tutorial-two-java.html
[http_www.rabbitmq.com_tutorials_tutorial-three-java.html]: http://www.rabbitmq.com/tutorials/tutorial-three-java.html
[http_www.rabbitmq.com_tutorials_tutorial-four-java.html]: http://www.rabbitmq.com/tutorials/tutorial-four-java.html
[http_www.rabbitmq.com_tutorials_tutorial-five-java.html]: http://www.rabbitmq.com/tutorials/tutorial-five-java.html