RabbitMQ——使用Exchange中的direct交换机实现消息发送和接收

package com.szh.rabbitmq.exchange.direct;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
/**
 *
 */
public class Send {
    public static void main(String[] args) {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.40.130");
        factory.setPort(5672);
        factory.setUsername("root");
        factory.setPassword("root");
        Connection connection=null;
        Channel channel=null;
        try {
            connection=factory.newConnection();
            channel=connection.createChannel();
            channel.queueDeclare("myDirectQueue",true,false,false,null);
            /**
             * 声明一个交换机
             * 参数1：交换机的名称，取值任意
             * 参数2：交换机的类型，取值为：direct、fanout、topic、headers
             * 参数3：是否为持久化的交换机
             * 注意：
             *      1) 声明交换机时，如果这个交换机已经存在，则放弃声明；如果交换机不存在，则声明该交换机
             *      2) 这行代码可有可无，但是使用前要确保该交换机已存在
             */
            channel.exchangeDeclare("directExchange","direct",true);
            /**
             * 将队列绑定到交换机
             * 参数1：队列的名称
             * 参数2：交换机的名称
             * 参数3：消息的RoutingKey（就是BindingKey）
             * 注意：
             *      1) 在进行队列和交换机绑定时，必须要确保交换机和队列已经成功声明
             */
            channel.queueBind("myDirectQueue","directExchange","directRoutingKey");
            /**
             * 发送消息到指定的队列
             * 参数1：交换机的名称
             * 参数2：消息的RoutingKey，如果该消息的RoutingKey和某个队列与交换机绑定的RoutingKey一致，则该消息会被发送到这个队列中
             * 注意：
             *      1) 发送消息时，必须确保交换机已经创建，同时已经正确的绑定到了某个队列上
             */
            String message="Exchange的direct消息绑定";
            channel.basicPublish("directExchange","directRoutingKey",null,message.getBytes(StandardCharsets.UTF_8));
            System.out.println("消息发送成功：" + message);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } finally {
            if (channel != null) {
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
            if (connection != null ) {
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

package com.szh.rabbitmq.exchange.direct;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
/**
 *
 */
public class Receive {
    public static void main(String[] args) {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.40.130");
        factory.setPort(5672);
        factory.setUsername("root");
        factory.setPassword("root");
        Connection connection=null;
        Channel channel=null;
        try {
            connection=factory.newConnection();
            channel=connection.createChannel();
            channel.queueDeclare("myDirectQueue",true,false,false,null);
            channel.exchangeDeclare("directExchange","direct",true);
            channel.queueBind("myDirectQueue","directExchange","directRoutingKey");
            //监听某个消息队列，同时获取消息队列中的数据
            channel.basicConsume("myDirectQueue",true,"",new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message=new String(body);
                    System.out.println("消息接收成功：" + message);
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}