RabbitMQ-订阅模式publish/subscribe

1.订阅模式模型

一个生产者多个消费者
每个消费者都有自己的队列
生产者没有直接把消息发送给队列，而是先发送给交换机exchange
每个队列都要绑定到交换机上
生产者发送的消息是经过交换机的，然后到达队列，就能实现一个消息被多个消费者消费

应用场景: 比如注册之后需要发送邮件同时需要发送短信

生产者

package com.ithzk.rabbitmq.ps;

import com.ithzk.rabbitmq.utils.RabbitMQConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/** * @author hzk * @date 2018/3/10 */
public class Send {

    private final static String EXCHANGER_NAME = "test_exchange_fanout";

    public static void main(String[] args) throws IOException, TimeoutException {

        Connection connection = RabbitMQConnectionUtils.getConnection();

        Channel channel = connection.createChannel();

        // 分发
        channel.exchangeDeclare(EXCHANGER_NAME,"fanout");

        String msg = "hello exchange";

        channel.basicPublish(EXCHANGER_NAME,"",null,msg.getBytes());

        System.out.println("Send msg"+msg);

        channel.close();
        connection.close();
    }

}

消息丢失了，因为交换机没有存储的能力，rabbitMQ中只有队列有存储能力，此时还没有队列绑定到该交换机上，所以数据丢失了。

消费者1

package com.ithzk.rabbitmq.ps;

import com.ithzk.rabbitmq.utils.RabbitMQConnectionUtils;
import com.rabbitmq.client.*;

import javax.sound.midi.Soundbank;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

/** * @author hzk * @date 2018/3/10 */
public class Recv1 {

    private static final String QUEUE_NAME="test_queue_fanout_email";
    private final static String EXCHANGER_NAME = "test_exchange_fanout";

    public static void main(String[] args) throws IOException, TimeoutException {
        //获取连接
        Connection connection = RabbitMQConnectionUtils.getConnection();

        //从连接中获取频道
        final Channel channel = connection.createChannel();

        //声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        //绑定队列到交换机 转发器
        channel.queueBind(QUEUE_NAME,EXCHANGER_NAME,"");

        //保证一次只发一个
        channel.basicQos(1);

        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("[1] Recv msg:" + msg);

                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("[1] done");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };

       boolean autoAck = false;

       channel.basicConsume(QUEUE_NAME,autoAck,consumer);

       System.out.println("[Consumer 1 start]");


    }


}

消费者2

package com.ithzk.rabbitmq.ps;

import com.ithzk.rabbitmq.utils.RabbitMQConnectionUtils;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/** * @author hzk * @date 2018/3/10 */
public class Recv2 {

    private static final String QUEUE_NAME="test_queue_fanout_sms";
    private final static String EXCHANGER_NAME = "test_exchange_fanout";

    public static void main(String[] args) throws IOException, TimeoutException {
        //获取连接
        Connection connection = RabbitMQConnectionUtils.getConnection();

        //从连接中获取频道
        final Channel channel = connection.createChannel();

        //声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        //绑定队列到交换机 转发器
        channel.queueBind(QUEUE_NAME,EXCHANGER_NAME,"");

        //保证一次只发一个
        channel.basicQos(1);

        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("[2] Recv msg:" + msg);

                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("[2] done");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };

       boolean autoAck = false;

       channel.basicConsume(QUEUE_NAME,autoAck,consumer);

       System.out.println("[Consumer 2 start]");


    }


}

和交换机绑定的队列都会收到消息