同步异步

同步

优点: 等待结果出来才返回, 实时性强 缺点:

• 可拓展性差 —> 耦合大
• 性能下降 —> 延时高
• 级联失败问题

异步调用

优点: 解除耦合, 拓展性强 无需等待,性能好 故障隔离 缓存消息/流量削峰填谷

缺点: 不能立即得到调用结果,时效性差 不确定下游业务执行是否成功 业务安全依赖于 Broker(消息队列) 的可靠性

flowchart LR
	消息发送者 ---> 消息队列 ---> 消息接收者1
	消息队列 ---> 消息接收者2

MQ

消息队列/字面来看就是存放消息的队列, 也就是异步调用中的 Broker 。

RabbitMQ

RabbitMQ 的整体架构及核心概念:

• virtual-host: 虚拟主机/起到数据隔离的作用
• publisher: 消息发送者
• consumer: 消息的消费者
• queue: 队列, 存储消息
• exchange: 交换机/负责路由消息

flowchart LR
    P[Publisher]

    subgraph Broker[RabbitMQ Server / Broker]
        subgraph VHost1[Virtual Host]
            E1[Exchange]
            E2[Exchange]

            Q1[Queue]
            Q2[Queue]
            Q3[Queue]
        end
    end

    C1[Consumer]
    C2[Consumer]

    P --> E1
    P --> E2

    E1 --> Q1

    E2 --> Q1
    E2 --> Q2
    E2 --> Q3

    Q1 --> C1
    Q2 --> C2
    Q3 --> C1

ex & queue

rabbitMQ中一定要将二者绑定, 直接给交换机发信息是没用的 -> not routed

visual host

实现多租户隔离,让不同的项目或业务线能在同一个 RabbitMQ 集群中互不干扰。 不同的 vhost 拥有各自独立的交换机、队列和权限体系,是天然的隔离屏障。

JAVA操作MQ

SpringAMQP

1. 引入依赖 spring-boot-starter-amqp

直接在父工程引入, 这样子工程可以省点

2.配置MQ地址

spring:
	rabbitmq:
		host:      #主机名
		port:
		virtual-host:   #虚拟机
		username: 
		password:

3.添加实例 RabbitTemplate

@Autowired
private RabbitTemplate rabbitTemplate;

public void test(){
	String queueName;
	String msg;
	rabbitTemplate.convertAndSend(queueName,msg);
}

4.收信息 @RabbitListener(queues=“name”)

@Component
public class test{
	@RabbitListener(queues="name") //注解写上要监听的队列
	public void listen(String msg) throws InterruptedException{
		...
	}
}

work queues

一种工作模型

• 多个消费者绑定到一个队列, 可以加快消息处理速度
• 同一条消息只会被一个消费者处理
• 通过设置prefetch来控制消费者预取的消息数量,处理完一条再处理下一条, 实现能者多劳

spring:
	rabbitmq:
		listener:
			simple:
				prefetch: 1  #每次只能获取一条信息, 处理完才获取下一跳
							 #否则会平均分配给消费者, 若处理能力弱, 会有明显堆积

交换机

Fanout 广播路由

1 → N 的扩散结构 Fanout Exchange 会将接收到的消息广播到每一个跟其绑定的 queue ,所以也叫广播模式

flowchart LR
	publisher ---> fanout-exchanges ---> queue1 ---> receiver1
	 fanout-exchanges ---> queue2 ---> receiver2

Direct 定向路由

• BindingKey不可拆分
• 每一个 Queue 都与 Exchange 设置一个 BindingKey
• 发布者发送消息时 , 指定消息的 RoutingKey
• Exchange 将消息路由到 BindingKey 与消息 RoutingKey一致的队列

flowchart LR
	publisher ---> direct-exchanges --key1--> queue1 ---> receiver1
	 direct-exchanges --key2--> queue2 ---> receiver2

Topic

• Topic 交换机接收的消息 RoutingKey可以是多个单词,以 .分割
• Topic 交换机与队列绑定时的bindingKey可以指定通配符
• #: 代表0个或者多个词
• *:代表1个词

flowchart LR
    publisher --> exchange[Topic Exchange]

    exchange -->|china.#| queue1[queue1]
    exchange -->|japan.#| queue2[queue2]
    exchange -->|#.weather| queue3[queue3]
    exchange -->|#.news| queue4[queue4]

    queue1 --> consumer1[consumer1]
    queue2 --> consumer2[consumer2]
    queue3 --> consumer3[consumer3]
    queue4 --> consumer4[consumer4]

声明队列

基于Bean太复杂, 此处略 以下是基于注解开发, @RabbitListener \ @QueueBinding \ @Queue \ @Exchange

	@RabbitListener(
			//绑定队列的属性: value, exchange, bindingkeys
			bindings= @QueueBinding(
				value=@Queue(name=""),
				exchange=@Exchange(name="", type=ExchangeTypes.DIRECT), //交换机名字和类型
				key={"","","",...}  //bindingkey数组
				)
	)public void method(){}

消息转换器

一些类型会序列化成字节

问题

Spring 的对消息对象的处理是由 org.springframework.amqp.support.converter.MessageConverter来 处理的。 而默认实现是 SimpIeMessageConverter, 基于 JDK 的 ObjectOutputStream 完成序列化

存在下列问题:

• JDK 的序列化有安全风险
• JDK 序列化的消息太大 ---> (jack, daidjoaijfojaofjpajfa)
• JDK 序列化的消息可读性差

解决

建议采刷 JSON 序列化 代替 默认的JDK 序列化 (SpringMVC底层的消息转换器就是jackson)

• 在 publisher 和 consumer 中都要引入 jackson 依赖:

<dependency>
	<group>com.fasterxml.jackson.core</groupId>
	<artifactId>jackson-databind<artifactId>
</dependency>

• 在 publisher 和 consumer 中都要配置 MessageConverter:

@Bean
public MessageConverter messageConverter(){
	return new Jackson2JsonMessageConverter();
}