zoukankan      html  css  js  c++  java
  • Rabbitmq(6) 主题模式

    * 匹配1个

    # 匹配所有

    发送者:

    package com.aynu.bootamqp.service;
    
    import com.aynu.bootamqp.commons.utils.Amqp;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Send {
    
        private final static String Exchange_NAME ="hello";
        public static void main(String[] args) throws IOException, TimeoutException {
                Connection connection = Amqp.getConnection();
                Channel channel = connection.createChannel();
                //声明交换机
                channel.exchangeDeclare(Exchange_NAME,"topic");
                //在手动确认机制之前
                //一次只发送一条消息,给不同的消费者
                channel.basicQos(1);
    
                String message = "hello ps";
                String routingKey ="goods.delete";
                channel.basicPublish(Exchange_NAME,routingKey,null,message.getBytes("utf-8"));
                System.out.println(message);
                channel.close();
                connection.close();
        }
    }

    接受者1

    package com.aynu.bootamqp.service;
    
    import com.aynu.bootamqp.commons.utils.Amqp;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    @SuppressWarnings("all")
    public class Receive {
    
        private final static String QUEUE_NAME ="hello";
        private final static String Exchange_NAME ="hello";
        public static void main(String[] args) throws IOException, TimeoutException {
            Connection connection = Amqp.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
            //绑定队列
    
            channel.queueBind(QUEUE_NAME,Exchange_NAME,"goods.add");
           // 一次只处理一个消息
            channel.basicQos(1);
            DefaultConsumer consumer = new DefaultConsumer(channel) {
    
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                                           AMQP.BasicProperties properties, byte[] body) throws IOException {
                    super.handleDelivery(consumerTag, envelope, properties, body);
                    String msg = new String(body,"utf-8");
                    System.out.println("receive"+msg);
                    try {
                        Thread.sleep(1000*2);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }finally {
                        // 手动发送消息确认机制
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    }
                }
            };
            // 自动应答
            boolean autoAck = false;
            channel.basicConsume(QUEUE_NAME,autoAck,consumer);
        }
    }

    接受者2

    package com.aynu.bootamqp.service;
    
    import com.aynu.bootamqp.commons.utils.Amqp;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    @SuppressWarnings("all")
    public class Receive2 {
    
        private final static String QUEUE_NAME ="hello1";
        private final static String Exchange_NAME ="hello";
        public static void main(String[] args) throws IOException, TimeoutException {
            Connection connection = Amqp.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
            channel.queueBind(QUEUE_NAME,Exchange_NAME,"goods.#");
            channel.basicQos(1);
            DefaultConsumer consumer = new DefaultConsumer(channel) {
    
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                                           AMQP.BasicProperties properties, byte[] body) throws IOException {
                    super.handleDelivery(consumerTag, envelope, properties, body);
                    String msg = new String(body,"utf-8");
                    System.out.println("receive2222"+msg);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }finally {
                        // 手动发送消息确认机制
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    }
                }
            };
            boolean autoAck = false;
            channel.basicConsume(QUEUE_NAME,autoAck,consumer);
        }
    }
  • 相关阅读:
    Windows 下Nexus搭建Maven私服
    WebService中获取request对象一例
    利用window.navigator.userAgent判断当前是否微信内置浏览器
    批量插入写法
    MySql 使用递归函数时遇到的级联删除问题
    【Mysql】 你会用 information_schema吗?
    Volatile 多线程中用到的关键字
    spring+springMVC中使用@Transcational方式管理事务的必须要配的东西。
    Android中操作SQLite数据库
    Oracle中的字符处理方法
  • 原文地址:https://www.cnblogs.com/mm163/p/10703709.html
Copyright © 2011-2022 走看看