微服务之五消息驱动_消息驱动 业务逻辑驱动

一： 概述

JavaEE提供了Message Driven Bean(消息驱动bean)，用于处理企业组件间的消息驱动。
Spring Cloud也引入了相应的驱动，Spring Cloud Stream
市面上存在的消息代理中间件

• ActivitiMQ
• RabbitMQ
• Kafka

类似邮局，有生产者、消息代理、消费者
Stream框架：

• 持久化订阅的支持
• 消费者组的支持
• Topic分区支持
• 支持RabbitMQ和Kafka两种消息代理组件

好处：不同考虑中间使用的什么代理机制，利用Stream实现消息的生产与发送
消息代理中间件模式

使用了Spring Cloud Stream结构后

使用了Stream后，生产者和消费者可以更加专注自己的业务，至于消息是如何投递、使用的那个消息代理则无需关心

二： RabbitMQ框架

RabbitMQ使用AMQP协议

2.1 消息生产者

消息的生产者/消费者都属于客户端，均使用AMQP协议与RabbitMQ服务器进行通信
添加依赖

<!-- AMQP -->
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>4.2.0</version>
</dependency>
<!-- SLF4J日志 -->
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>1.7.9</version>
</dependency>
1
2
3
4
5
6
7
8
9
10
11
12

发消息

package com.atm.cloud;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class SendMessage {
   

    public static void main(String[] args) throws Exception {
   
        /*
         * 1.生产者会发送消息给RabbitMQ服务器。 2.通过渠道叫消息发送给交换器。 3.交换器会发送给队列。 4.队列将消息发送给消费者。
         */

        // 建立连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 设置host，其实无需设置，默认为localhost，用户名/密码默认guest，端口默认5671
        // factory.setHost("localhost");

        // 创建新的连接
        Connection connection = factory.newConnection();

        // 通过连接创建渠道（向该渠道发送消息）
        Channel channel = connection.createChannel();

        // 声明交换器(默认绑定)，交换器会将消息发送给队列，对列再发送给消费者
        // 直接声明队列，使用默认交换器
        String queueName = "MyQueueName";

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

        // 创建消息，使用渠道发布消息，""使用默认交换器，本列子中routingKey就使用queueName
        String messageBody = "Hello Wrold!";
        channel.basicPublish("", queueName, null, messageBody.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

2.2 消费者

package com.atm.cloud;

import java.io.IOException;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class ReadMessage {
   

    public static void main(String[] args) throws Exception {
   

        // 建立连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 创建新的连接
        Connection connection = factory.newConnection()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22