rabbitMQ简单的生产、消费者模式-----direct(路由完全匹配)

- 日理万妓 2022-12-22 04:49 37阅读 0赞

生产者代码

package cn.enjoyedu.myexchange.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;
    
    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();
            //设置工厂的地址
            connectionFactory.setHost("localhost");
            //创建连接
            Connection connection = connectionFactory.newConnection();
            //创建信道
            Channel channel = connection.createChannel();
            //设置信道的交换器
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
            //声明路由，消息体
            String[] routeKeys = {"meimei","nannan","someBody"};
            for (int i = 0; i < routeKeys.length; i++) {
                String routekey = routeKeys[i%3];
                String mes = "hello rabbitmq "+ routekey+(i+1);
                //发布消息
                channel.basicPublish(EXCHANGE_NAME,routekey,null,mes.getBytes());
            }
            //关闭资源
            channel.close();
            connection.close();
    
        }
    }

消费者代码

package cn.enjoyedu.myexchange.direct;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class NormalConsumer {
        public static void main(String[] args) throws IOException, TimeoutException {
            //创建连接的工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //设置连接mq的地址
            connectionFactory.setHost("localhost");
            //创建连接
            Connection connection = connectionFactory.newConnection();
            //创建信道
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
            //声明队列
            String queueName = "queue-meimei";
            channel.queueDeclare(queueName,false,false,false,null);
            System.out.println("waiting for message ......");
    
            //声明消费者
            final Consumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String routingKey = envelope.getRoutingKey();
                    String mes = new String(body,"utf-8");
                    System.out.println("routingKey["+routingKey+"] "+mes);
    
                }
            };
            //消息者正是开始在指定队列上消费。(queue-king)
            //TODO 这里第二个参数是自动确认参数，如果是true则是自动确认
            channel.basicConsume(queueName,true,consumer);
    
    
        }
    }