rabbitmq队列和交换器的多重绑定、一个连接多个信道、个队列多个消费者，则会表现出消息在消费者之间的轮询发送。

红太狼 2022-12-22 04:49 108阅读 0赞

生产者

package cn.enjoyedu.exchange.direct;
    
    
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     *类说明：direct类型交换器的生产者
     */
    public class DirectProducer {
    
        public final static String EXCHANGE_NAME = "direct_logs";
    
        public static void main(String[] args)
                throws IOException, TimeoutException {
            //创建连接、连接到RabbitMQ
            ConnectionFactory connectionFactory= new ConnectionFactory();
            //设置下连接工厂的连接地址(使用默认端口5672)
            connectionFactory.setHost("localhost");
            //创建连接
            Connection connection =connectionFactory.newConnection();
            //创建信道
            Channel channel =connection.createChannel();
    
            //在信道中设置交换器
            channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.DIRECT);
    
            //申明队列（放在消费者中去做）
            //申明路由键\消息体
            String[] routeKeys ={"king","mark","james"};
            for (int i=0;i<6;i++){
                String routeKey = routeKeys[i%3];
                String msg = "Hello,RabbitMQ"+(i+1);
                //发布消息
                channel.basicPublish(EXCHANGE_NAME,routeKey,null,msg.getBytes());
                System.out.println("Sent:"+routeKey+":"+msg);
            }
            channel.close();
            connection.close();
    
        }
    
    }

消费者：队列和交换器的多重绑定

package cn.enjoyedu.exchange.direct;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     *类说明：队列和交换器的多重绑定
     */
    public class MultiBindConsumer {
    
        public static void main(String[] argv) throws IOException,
                InterruptedException, TimeoutException {
            //连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            //连接rabbitMq的地址
            factory.setHost("127.0.0.1");
    
            // 打开连接和创建频道，与发送端一样
            Connection connection = factory.newConnection();
            //创建一个信道
            final Channel channel = connection.createChannel();
            //信道设置交换器类型(direct)
            channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME,BuiltinExchangeType.DIRECT);
    
            //声明一个随机队列
            String queueName = channel.queueDeclare().getQueue();
            //TODO
            /*队列绑定到交换器上时，是允许绑定多个路由键的，也就是多重绑定*/
            String[] routekeys={"king","mark","james"};
            for(String routekey:routekeys){
                channel.queueBind(queueName,DirectProducer.EXCHANGE_NAME,
                        routekey);
            }
            System.out.println(" [*] Waiting for messages:");
    
            // 创建队列消费者
            final Consumer consumerA = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties
                                                   properties,
                                           byte[] body)
                        throws IOException {
                    String message = new String(body, "UTF-8");
                    System.out.println(" Received "
                            + envelope.getRoutingKey() + ":'" + message
                            + "'");
                }
            };
            channel.basicConsume(queueName, true, consumerA);
        }
    }

消费者：一个连接多个信道

package cn.enjoyedu.exchange.direct;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     *类说明：一个连接多个信道
     */
    public class MultiChannelConsumer {
    
        private static class ConsumerWorker implements Runnable{
            final Connection connection;
    
            public ConsumerWorker(Connection connection) {
                this.connection = connection;
            }
    
            public void run() {
                try {
                    /*创建一个信道，意味着每个线程单独一个信道*/
                    Channel channel = connection.createChannel();
                    //信道设置交换器类型(direct)
                    channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME,BuiltinExchangeType.DIRECT);
                    // 声明一个随机队列
                     String queueName = channel.queueDeclare().getQueue();
                    //String queueName = "queue-king";      // 同一个队列
    
                    //消费者名字，打印输出用
                    final String consumerName =  Thread.currentThread().getName()+"-all";
    
                    /*队列绑定到交换器上时，是允许绑定多个路由键的，也就是多重绑定*/
                    String[] routekeys={"king","mark","james"};
                    for(String routekey:routekeys){
                        channel.queueBind(queueName,DirectProducer.EXCHANGE_NAME,
                                routekey);
                    }
                    System.out.println("["+consumerName+"] Waiting for messages:");
    
                    // 创建队列消费者
                    final Consumer consumerA = new DefaultConsumer(channel) {
                        @Override
                        public void handleDelivery(String consumerTag,
                                                   Envelope envelope,
                                                   AMQP.BasicProperties
                                                           properties,
                                                   byte[] body)
                                throws IOException {
                            String message =
                                    new String(body, "UTF-8");
                            System.out.println(consumerName
                                    +" Received "  + envelope.getRoutingKey()
                                    + ":'" + message + "'");
                        }
                    };
                    channel.basicConsume(queueName, true, consumerA);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    
        public static void main(String[] argv) throws IOException,
                InterruptedException, TimeoutException {
            //连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            //连接rabbitMq的地址
            factory.setHost("127.0.0.1");
            // 打开连接和创建频道，与发送端一样
            Connection connection = factory.newConnection();
            //一个连接多个信道
            for(int i=0;i<2;i++){
                /*将连接作为参数，传递给每个线程*/
                Thread worker =new Thread(new ConsumerWorker(connection));
                worker.start();
            }
        }
    }

消费者：一个队列多个消费者，则会表现出消息在消费者之间的轮询发送

package cn.enjoyedu.exchange.direct;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     *类说明：一个队列多个消费者，则会表现出消息在消费者之间的轮询发送。
     */
    public class MultiConsumerOneQueue {
    
        private static class ConsumerWorker implements Runnable{
            final Connection connection;
            final String queueName;
    
            public ConsumerWorker(Connection connection,String queueName) {
                this.connection = connection;
                this.queueName = queueName;
            }
    
            public void run() {
                try {
                    /*创建一个信道，意味着每个线程单独一个信道*/
                    final Channel channel = connection.createChannel();
                    //信道设置交换器类型(direct)
                    channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME,BuiltinExchangeType.DIRECT);
                    /*声明一个队列,rabbitmq，如果队列已存在，不会重复创建*/
                    channel.queueDeclare(queueName, false,false, false,null);
                    //消费者名字，打印输出用
                    final String consumerName =  Thread.currentThread().getName();
    
                    /*队列绑定到交换器上时，是允许绑定多个路由键的，也就是多重绑定*/
                    String[] routekeys={"king","mark","james"};
                    for(String routekey:routekeys){
                        channel.queueBind(queueName,DirectProducer.EXCHANGE_NAME,
                                routekey);
                    }
                    System.out.println(" ["+consumerName+"] Waiting for messages:");
    
                    // 创建队列消费者
                    final Consumer consumerA = new DefaultConsumer(channel) {
                        @Override
                        public void handleDelivery(String consumerTag,
                                                   Envelope envelope,
                                                   AMQP.BasicProperties
                                                           properties,
                                                   byte[] body)
                                throws IOException {
                            String message =
                                    new String(body, "UTF-8");
                            System.out.println(consumerName
                                    +" Received "  + envelope.getRoutingKey()
                                    + ":'" + message + "'");
                        }
                    };
                    channel.basicConsume(queueName, true, consumerA);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    
        public static void main(String[] argv) throws IOException,
                InterruptedException, TimeoutException {
            //连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            //连接rabbitMq的地址
            factory.setHost("127.0.0.1");
    
            // 打开连接和创建频道，与发送端一样
            Connection connection = factory.newConnection();
            //TODO
            //3个线程，线程之间共享队列,一个队列多个消费者
            String queueName = "focusAll";
            for(int i=0;i<3;i++){
                /*将队列名作为参数，传递给每个线程*/
                Thread worker =new Thread(new ConsumerWorker(connection,queueName));
                worker.start();
            }
    
        }
    }