RabbitMQ队列和交换机类型_rabbmit mq的队列模式和交换机模式

队列

1简单队列

• P：生产者
• 红色：队列Q
• C：消费者

不足：

• 耦合性高，生产者消费者一一对应，多个消费者不能消费同一个队列，队列名也需要同时变更。

1.1 生产者Productor

工具类，获取mq连接

/**
* 获取MQ连接
 * @return
 * @throws IOException
 * @throws TimeoutException
 */
public static Connection getConnection() throws IOException, TimeoutException {
   
    //定义工厂
    ConnectionFactory factory = new ConnectionFactory();
    //服务地址
    factory.setHost("127.0.0.1");
    //AMQP端口
    factory.setPort(5672);
    //vhost
    factory.setVirtualHost("/");
    //用户密码
    factory.setUsername("guest");
    factory.setPassword("guest");

    return factory.newConnection();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

private static final String QUEUE_NAME="test_simple_queue";
public static void main(String[] args) throws IOException, TimeoutException {
   
    Connection connection = ConnectionUtils.getConnection();

    //从连接中获取通道
    Channel channel = connection.createChannel();
    //创建队列
    channel.queueDeclare(QUEUE_NAME,false,false,false,null);

    //发布消息
    String msg = "hello simple";
    channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());

    channel.close();
    connection.close();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

出现队列

队列进行Get Message（s）取出消息，Message变为0。

1.2 消费者Consumer

public class Consumer {
   
    private static final String QUEUE_NAME="test_simple_queue";
    public static void main(String[] args) throws IOException, TimeoutException {
   
        //创建连接
        Connection connection = ConnectionUtils.getConnection();
        //创建频道
        Channel channel = connection.createChannel();

        //队列声明（生产者中声明过可以不写）
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        //消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
   
            //事件模型，一旦有消息进入队列即触发方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
   
                String msg = new String(body, "utf-8");
                System.out.println("接收到的消息：" + msg);
            }
        };

        //监听队列
        channel.basicConsume(QUEUE_NAME
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26