RabbitMQ-Java实现Publish/Subscribe订阅模式

淩亂°似流年 2022-11-13 10:25 140阅读 0赞

*  订阅模式

一个生成者，多个消费者，每个消费者有自己的队列，生产者没有直接把消息发到队列，而是发给了交换机exchange

适合场景举例：对于同一个消息，要发邮件，也要发短信，因此拆分成两个队列

*  新建连接RabbitMQ的工具类utils

import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
     
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
     
    public class utils {
        public  static Connection getConnection() throws IOException, TimeoutException {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            factory.setPort(5672);
            factory.setVirtualHost("/vhost_zdy");
            factory.setUsername("username");
            factory.setPassword("password");
     
            return factory.newConnection();
        }
    }

*  生产者类PubSubSend

import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
     
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
     
    public class PubSubSend {
        private final static String EXCHANGE_NAME = "PubSub_exchange";
        public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
            Connection connection = utils.getConnection();
            Channel channel = connection.createChannel();
            //声明交换机
            channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
     
            for(int i=0;i<50;i++)//发50条消息
            {
                String msg="msg"+i;
                channel.basicPublish(EXCHANGE_NAME, "", null, msg.getBytes());
                Thread.sleep(20);//假设每20ms发送一次消息
                System.out.println("send:"+ msg);
            }
     
            channel.close();
            connection.close();
     
        }
    }

运行后在控制台exchange查看是否生成交换机

*  消费者PubSubReceive\_1和PubSubReceive\_2

import com.rabbitmq.client.*;
     
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
     
    public class PubSubReceive_1 {
        private final static String QUEUE_NAME = "PubSub_queue_mail";
        private final static String EXCHANGE_NAME = "PubSub_exchange";
        public static void main(String[] args) throws IOException, TimeoutException {
            Connection connection = utils.getConnection();
            final Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"");
            channel.basicQos(1);//保证一次只分发一个\n" +
            System.out.println(" [1] Waiting for messages.");
     
            DefaultConsumer defaultConsumer= 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("[1]receive:"+ msg);
                    try {
                        Thread.sleep(2000);//假设消费者1处理时间为2s
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }finally {
                        System.out.println("[1]done:"+ msg);
                        //手动回执
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    }
                }
            };
            boolean autoack=false;
            channel.basicConsume(QUEUE_NAME,autoack,defaultConsumer);
        }
    }
    import com.rabbitmq.client.*;
     
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
     
    public class PubSubReceive_2 {
        private final static String QUEUE_NAME = "PubSub_queue_tel";
        private final static String EXCHANGE_NAME = "PubSub_exchange";
        public static void main(String[] args) throws IOException, TimeoutException {
            Connection connection = utils.getConnection();
            final Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"");
            channel.basicQos(1);//保证一次只分发一个\n" +
            System.out.println(" [2] Waiting for messages.");
     
            DefaultConsumer defaultConsumer= 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("[2]receive:"+ msg);
                    try {
                        Thread.sleep(1000);//假设消费者1处理时间为2s
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }finally {
                        System.out.println("[2]done:"+ msg);
                        //手动回执
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    }
                }
            };
            boolean autoack=false;
            channel.basicConsume(QUEUE_NAME,autoack,defaultConsumer);
        }
    }