RabbitMQ常用的有6种工作模式，它们分别是简单模式，work模式，发布订阅模式，路由模式，Topic模式。下面就分别解释一下这些工作模式的特点和实现方式。

简单模式

lHiZhF.png
简单模式的构成由一个消息生产者，一个消息消费者和一个队列组成。

它的代码实现如下：
消息生产者的实现方式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public class Send {

    private static String queueName="simple_queue";
    public static void main(String[] args) throws Exception {

        //这里使用了我们自己编写的一个获取连接的工具类
        final Connection connection = ConnectionUtils.getConnection();
        //从连接中拿到一个通道对象
        final Channel channel = connection.createChannel();

        //声明一个队列
        channel.queueDeclare(queueName,false,false,false,null);

        //向队列中发送消息
        channel.basicPublish("",queueName,null,"hello world".getBytes());

        //关闭通道和连接
        channel.close();
        connection.close();

    }
}

消息消费者的实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public class Resv {
    private static String queueName="simple_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        final Connection connection = ConnectionUtils.getConnection();
        final Channel channel = connection.createChannel();

        //创建一个消费者
        final DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("收到了消息"+new String(body));
            }
        };
        //第一参数是队列名称，第二个参数指开启自动应答，第三个为消费者
        channel.basicConsume(queueName,true,consumer);
    }


}

work模式

lHVvt0.png
一个消息生产者多个消息消费者，每个消费者获取到的消息都是唯一的。

消息生产者的实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
public class Send {

    private static String queueName="simple_queue";
    public static void main(String[] args) throws Exception {

        final Connection connection = ConnectionUtils.getConnection();
        //从连接中拿到一个通道对象
        final Channel channel = connection.createChannel();

        //声明一个队列
        channel.queueDeclare(queueName,false,false,false,null);

        for(int i=0;i<100;i++){
            //向队列中发送消息
            channel.basicPublish("",queueName,null,(i+" hello world").getBytes());
            Thread.sleep(500);
        }
        //关闭通道和连接
        channel.close();
        connection.close();

    }
}

两个消息消费者的实现如下：

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

public class Resv {
    private static String queueName="simple_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        final Connection connection = ConnectionUtils.getConnection();
        final Channel channel = connection.createChannel();

        //创建一个消费者
        final DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("接收者1收到了消息"+new String(body));
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        channel.basicConsume(queueName,true,consumer);
    }
}



public class Resv2 {
    private static String queueName="simple_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        final Connection connection = ConnectionUtils.getConnection();
        final Channel channel = connection.createChannel();

        //创建一个消费者
        final DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("接收者2收到了消息"+new String(body));
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        channel.basicConsume(queueName,true,consumer);
    }
}

如果在消息消费者处理消息的时候，消息消费者下线了，那么以上这种模式就会导致消息丢失。我们可以通过消息应当机制来处理这种问题。
带有手动消息确认机制的消息消费者的实现如下：

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
27
28
29
30
31

public class Resv {
    private static String queueName="simple_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        final Connection connection = ConnectionUtils.getConnection();
        final Channel channel = connection.createChannel();

        channel.basicQos(1);

        //创建一个消费者
        final DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("接收者1收到了消息"+new String(body));
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                //手动发送回执，完成任务之后，手动确认消息
                channel.basicAck(envelope.getDeliveryTag(),false);
            }

        };
        //第二个参数为false，关闭自动应当
        channel.basicConsume(queueName,false,consumer);
    }
}

订阅模式

lHegII.png
发布订阅模式，拥有一个消息生产者，交换器，多个消息队列。
消息生产者的实现代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

public class Send {
    private static final String EXCHANGE_NAME="test_exchange_fanout";
    public static void main(String[] args) throws IOException, TimeoutException {
        final Connection connection = ConnectionUtils.getConnection();
        final Channel channel = connection.createChannel();

        //定义一个交换机，fanout为交换类型，代表不处理路由键
        //fanout交换类型，会把消息推送到所有的队列中去
        channel.exchangeDeclare(EXCHANGE_NAME,"fanout");

        for (int i=0;i<100;i++){
            //将消息推送到交换机上，再由交换机将消息转发到相应的消息队列中去，交换机不具备存储能力
            channel.basicPublish(EXCHANGE_NAME,"",null,(i+" 消息").getBytes());
        }
        channel.close();
        connection.close();

    }
}

消息消费者的代码实现：

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
27
28
29
30
31
32
33
34
public class Recv1 {

    private static final String QUEUE_NAME="test_queue1";
    private static final String EXCHANGE_NAME="test_exchange_fanout";
    public static void main(String[] args) throws IOException, TimeoutException {
        final Connection connection = ConnectionUtils.getConnection();
        final Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        //将队列绑定到，交换机上
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"");

        channel.basicQos(1);
        final DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("收到消息" + new String(body, "utf-8"));

                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }

        };
        //第二个参数关闭自动应答
        channel.basicConsume(QUEUE_NAME,false,consumer);

    }
}

另一个消息消费者的代码(除了声明了另一个队列之外，没有任何的特别之处)：

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
27
28
29
30
31
32
33
34
35

public class Recv2 {

    private static final String QUEUE_NAME="test_queue2";
    private static final String EXCHANGE_NAME="test_exchange_fanout";
    public static void main(String[] args) throws IOException, TimeoutException {
        final Connection connection = ConnectionUtils.getConnection();
        final Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        //绑定到转发器
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"");

        channel.basicQos(1);
        final DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("收到消息" + new String(body, "utf-8"));

                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }

        };
        //第二个参数关闭自动应答
        channel.basicConsume(QUEUE_NAME,false,consumer);
    }
}

路由模式

lHB4pV.png
路由模式和发布订阅模式非常的类似，只不过路由模式的交换器会处理路由键。
消息生产者的实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public class Send {

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

        final Channel channel = connection.createChannel();
        
        //交换机的工作模式设置为direct模式，它会处理路由键
        channel.exchangeDeclare(EXCHANGE_NAME,"direct");

        for(int i=0;i<100;i++){
            //路由键为info
            channel.basicPublish(EXCHANGE_NAME,"info",null,(i+"info消息").getBytes());
        }

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

消息接收者的实现如下；

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
//接收者1
public class Recv {

    private static final String EXCHANGE_NAME="test_exchange_direct";


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

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);


        //该队列指接收路由键为error的
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"error");
        
        channel.basicQos(1);
        final DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("收到消息" + new String(body, "utf-8"));

                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }

        };
        //第二个参数关闭自动应答
        channel.basicConsume(QUEUE_NAME,false,consumer);

    }
}

//接收者2

public class Recv2 {
    private static final String EXCHANGE_NAME="test_exchange_direct";

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

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);



        //该队列会接收以下三种路由键的
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"error");
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"info");
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"warning");

        channel.basicQos(1);

        final DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("收到消息" + new String(body, "utf-8"));

                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }

        };
        //第二个参数关闭自动应答
        channel.basicConsume(QUEUE_NAME,false,consumer);

    }
}

Topic模式

lHrp80.png
Topic会根据路由键和模式的匹配来决定，将消息发送到哪个队列中去。

消息生产者的实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public class Send {
    
    private static final String EXCHANGE_ANME="test_exchange_topic";
    public static void main(String[] args) throws IOException, TimeoutException {
        final Connection connection = ConnectionUtils.getConnection();
        final Channel channel = connection.createChannel();

        //将交换机设置为topic模式
        channel.exchangeDeclare(EXCHANGE_ANME,"topic");

        for(int i=0;i<100;i++){
            channel.basicPublish(EXCHANGE_ANME,"goods",null,(i+"消息").getBytes());
        }
        channel.close();
        connection.close();
        

    }
}

消息消费者的实现如下：

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
//消息消费者A
public class Recv {

    private static final String EXCHANGE_NAME="test_exchange_topic";


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

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);


        //‘*’表示子匹配一个词，而‘#’表示匹配多个词，词之间使用“.‘分隔
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"goods.*");
        channel.basicQos(1);
        final DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("收到消息" + new String(body, "utf-8"));

                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }

        };
        //第二个参数关闭自动应答
        channel.basicConsume(QUEUE_NAME,false,consumer);

    }
}

//消息消费者B
public class Recv2 {
    private static final String EXCHANGE_NAME="test_exchange_topic";

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

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);



        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"goods.add");


        channel.basicQos(1);

        final DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("收到消息" + new String(body, "utf-8"));

                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }

        };
        //第二个参数关闭自动应答
        channel.basicConsume(QUEUE_NAME,false,consumer);

    }
}

技术RabbitMQ
许可协议

本文采用 署名-非商业性使用-相同方式共享 4.0 国际 许可协议，转载请注明出处。

快来参与讨论吧~