阿里云AMQP与SpringBoot配置实践

小咪咪 2023-03-13 02:10 64阅读 0赞

**规范要求**

*  命名规范：生产者、消费者、事件、消息 定义
 *  生产和非生产 账户隔离；   
 *  virtual-host区分开发dev、生产prod、测试test （virtual-host支持最多10个，项目多了支持不了）
 *  针对某些场景发送量较大，如日志、短信、邮件等非核心业务服务消息，生产可独立账户或virtual-host（避免影响业务数据）
 *  exchange  路由命名： **路由类型.生产者名**( amq.topic.oa )  ，阿里云支持3中类型**（direct【 Binding Key = Routing Key ，适用于区分发往同一个 Exchange】，topic【 Binding Key 和 Routing Key 的通配符匹配路由消息至 Queue】，fanout【将消息路由至所有绑定的 Queue】）**
 *  routing key 命名：    **生产者名.事件名**； 
 *  queue 队列命名：**消费者名.模型名.生产者名； **
 *  bindingkey：**  生产者名.事件名（=routingkey）；**
 *  **发送端注意发送内容相同，但事件、routing key不一样；****消费者消费一样的内容，但有多个事件需要进行绑定；**

**Spring Boot 配置**

在`pom.xml`中引入如下依赖内容

<table> 
 <tbody> 
 <tr> 
 <td> <code>&lt;dependency&gt;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>&lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>&lt;artifactId&gt;spring-boot-starter-amqp&lt;/artifactId&gt;</code> <code>&lt;/dependency&gt;</code> <code>&lt;dependency&gt;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>&lt;groupId&gt;com.alibaba.mq-amqp&lt;/groupId&gt;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>&lt;artifactId&gt;mq-amqp-client&lt;/artifactId&gt;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>&lt;version&gt;1.0.4&lt;/version&gt;</code> <code>&lt;/dependency&gt;</code> </td> 
 </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
 <tr> 
 <td> <code>spring.rabbitmq.host=1618060924840827.mq-amqp.cn-shenzhen-a.aliyuncs.com&nbsp;</code><code>#地址</code> <code>spring.rabbitmq.port=5672&nbsp;</code><code>#端口</code> <code>spring.rabbitmq.username=LTAI4FchjfDJ5L54MRr6DngE 用户名</code> <code>spring.rabbitmq.password=****** 密码</code> <code>spring.rabbitmq.virtual-host=hscf-oa 连接到AMQP的那个vhost&nbsp;</code> <code>spring.rabbitmq.exchange-name=amq.topic Exchang名称 会有多种Exchang.</code> <code>#-----发送方的确认</code> <code>spring.rabbitmq.publisher-confirms=</code><code>true</code>&nbsp;&nbsp;&nbsp;<code># 是否启用【发布确认】</code> <code>spring.rabbitmq.publisher-returns=</code><code>true</code>&nbsp;&nbsp;&nbsp;&nbsp;<code># 是否启用【发布返回】</code> <code>spring.rabbitmq.template.mandatory=</code><code>true</code>&nbsp;&nbsp;&nbsp;<code># 触发returnedMessage回调必须设置mandatory=true, 否则Exchange没有找到Queue就会丢弃掉消息, 而不会触发回调</code> <code>#-------------消费端配置</code> <code>spring.rabbitmq.listener.simple.acknowledge-mode=manual&nbsp;</code><code># 设置手动确认(ack) Queue -&gt; C 消费端配置</code> <code>spring.rabbitmq.listener.simple.concurrency=1&nbsp;</code><code>#最小的消费者数量</code> <code>spring.rabbitmq.listener.simple.max-concurrency=5&nbsp;</code><code>#最大的消费者数量</code> <code># spring.rabbitmq.listener.simple.default-requeue-rejected=true #消费者消费失败，自动重新入队</code> </td> 
 </tr> 
 </tbody> 
</table>

配置权限认证相关类，创建一个 AliyunCredentialsProvider类实现CredentialsProvider类

<table> 
 <tbody> 
 <tr> 
 <td> 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 </td> 
 <td> <code>/**</code> <code>&nbsp;</code><code>* @author wangmingsen</code> <code>&nbsp;</code><code>* @create 2019-09-21-10:17</code> <code>&nbsp;</code><code>**/</code> &nbsp; <code>public</code>&nbsp;<code>class</code>&nbsp;<code>AliyunCredentialsProvider&nbsp;</code><code>implements</code>&nbsp;<code>CredentialsProvider { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>/**</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>* Access Key ID.</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*/</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>final</code>&nbsp;<code>String accessKeyId;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>/**</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>* Access Key Secret.</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*/</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>final</code>&nbsp;<code>String accessKeySecret;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>/**</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>* security temp token. (optional)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*/</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>final</code>&nbsp;<code>String securityToken;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>/**</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>* 资源owner账号（主账号）</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*/</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>final</code>&nbsp;<code>long</code>&nbsp;<code>resourceOwnerId;</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>public</code>&nbsp;<code>AliyunCredentialsProvider(</code><code>final</code>&nbsp;<code>String accessKeyId,&nbsp;</code><code>final</code>&nbsp;<code>String accessKeySecret,</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>final</code>&nbsp;<code>long</code>&nbsp;<code>resourceOwnerId) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>(accessKeyId, accessKeySecret,&nbsp;</code><code>null</code><code>, resourceOwnerId);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>public</code>&nbsp;<code>AliyunCredentialsProvider(</code><code>final</code>&nbsp;<code>String accessKeyId,&nbsp;</code><code>final</code>&nbsp;<code>String accessKeySecret,</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>final</code>&nbsp;<code>String securityToken,&nbsp;</code><code>final</code>&nbsp;<code>long</code>&nbsp;<code>resourceOwnerId) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.accessKeyId = accessKeyId;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.accessKeySecret = accessKeySecret;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.securityToken = securityToken;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>this</code><code>.resourceOwnerId = resourceOwnerId;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>@Override</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>public</code>&nbsp;<code>String getUsername() { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code><code>(StringUtils.isNotEmpty(securityToken)) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code>&nbsp;<code>UserUtils.getUserName(accessKeyId, resourceOwnerId, securityToken);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}&nbsp;</code><code>else</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code>&nbsp;<code>UserUtils.getUserName(accessKeyId, resourceOwnerId);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>@Override</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>public</code>&nbsp;<code>String getPassword() { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>try</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code>&nbsp;<code>UserUtils.getPassord(accessKeySecret);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}&nbsp;</code><code>catch</code>&nbsp;<code>(InvalidKeyException e) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//todo</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}&nbsp;</code><code>catch</code>&nbsp;<code>(NoSuchAlgorithmException e) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//todo</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code>&nbsp;<code>null</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

创建消息生产者,通过注入`RabbitTemplate`实例来实现消息的发送

创建消费者来消费消息.。通过`@RabbitListener`注解定义该类对Queue队列的监听,通过`@RabbitHandler`注解来指定对消息的处理方法

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 </td> 
 <td> <code>RabbitHandler</code> <code>@RabbitListener</code><code>(queues = { </code><code>"${spring.rabbitmq.queue.after-loan}"</code><code>})</code> <code>public</code>&nbsp;<code>void</code>&nbsp;<code>process(Message message, Channel channel)&nbsp;</code><code>throws</code>&nbsp;<code>IOException { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>try</code><code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>String carInfo =&nbsp;</code><code>new</code>&nbsp;<code>String(message.getBody(), StandardCharsets.UTF_8);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>log.info(</code><code>"【MQ消息接收】xxx信息: {}"</code><code>, carInfo);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>Car car = JSON.parseObject(carInfo, Car.</code><code>class</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>channel.basicAck(message.getMessageProperties().getDeliveryTag(),</code><code>false</code><code>);&nbsp;</code><code>//确认消费</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}&nbsp;</code><code>catch</code>&nbsp;<code>(Exception e) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>log.error(</code><code>"【MQ消息接收】xxxx处理异常"</code><code>, e);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);&nbsp; //丢弃这条消息</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

下面是几种Exchange类型，我们目前使用的是Topic Exchange.

### Exchange 类型 ###

本文介绍消息队列 AMQP 中不同类型的 Exchange 的路由规则及其使用场景，帮助您更好地根据使用场景选择合适的 Exchange 类型。

背景信息

Producer 将消息发送至 Exchange。Exchange 根据 Routing Key 和 Binding Key 路由消息。Routing Key 是 Producer 为消息添加的属性。Binding Key 是 Exchange 和 Queue 之间的绑定关系。不同类型的 Exchange 在路由消息时的规则不同，适用于不同的使用场景。

### Direct Exchange ###

路由规则

Direct Exchange 将消息路由至 Binding Key 与消息的 Routing Key 完全匹配的 Queue。

使用场景

Direct Exchange 根据 Binding Key = Routing Key 的规则路由消息，适用于区分发往同一个 Exchange 的消息。

示例

**图 1. Direct Exchange[![format_png][]][format_png 1]**

如[图 1][1] 所示，Direct Exchange 分别与 Queue A 和 Queue B 绑定。Consumer 可以从 Queue A 中拉取所有图片（img）创建（create）消息，从 Queue B 中拉取所有图片（img）日志（log）消息。  
Producer 向 Direct Exchange 发送消息：

*  消息 A 的 Routing Key 为 img.create。
 *  消息 B 的 Routing Key 为 img.log。

**Direct Exchange 绑定 Queue：**

*  Direct Exchange 和 Queue A 的 Binding Key 为 img.create。
 *  Direct Exchange 和 Queue B 的 Binding Key 为 img.log。

**Direct Exchange 根据 Binding Key = Routing Key 的规则路由消息：**

*  Routing Key 为 img.create 消息 A 路由至 Binding Key 为 img.create 的 Queue A。
 *  Routing Key 为 img.log 消息 B 路由至 Binding Key 为 img.log 的 Queue B。

### Topic Exchange ###

路由规则

Topic Exchange 根据 Binding Key 和 Routing Key 的通配符匹配路由消息至 Queue。

说明

*  星号（\*）代表一个单词（例如，cn）。
 *  井号（\#）代表零个或多个单词（例如，cn.zj.hz）。

使用场景

Topic Exchange 使用通配符匹配的路由模式，适用于通过 Topic 对消息进行区分的场景。

示例

**图 2. Topic Exchange[![format_png 2][]][format_png 2 1]**

如[图 2][2] 所示，Topic Exchange 分别与 Queue A 和 Queue B 绑定。Consumer 可以从 Queue A 中拉取所有中国（cn）杭州（hz）的文件（files），从 Queue B 中拉取所有中国（cn）商店（store）的文件（files）。  
Producer 向 Topic Exchange 发送消息：

*  消息 A 的 Routing Key 为 [files.cn][].hz。
 *  消息 B 的 Routing Key 为 [files.cn][].hz.store。
 *  消息 C 的 Routing Key 为 [files.cn.sz][].store。

**Topic Exchange 绑定 Queue：**

*  Topic Exchange 和 Queue A 的 Binding Key 为 [files.cn][].hz.\#。
 *  Topic Exchange 和 Queue B 的 Binding Key 为 [files.cn][].\*.store。

**Topic Exchange 根据通配符匹配的规则路由消息：**

*  Routing Key 为 [files.cn][].hz 消息 A 路由至 Binding Key 为 [files.cn][].hz.\# 的 Queue A。
 *  Routing Key 为 [files.cn][].hz.store 消息 B 路由至 Binding Key 为 [files.cn][].hz.\# 的 Queue A 和 Binding Key 为 [files.cn][].\*.store 的 Queue B。
 *  Routing Key 为 [files.cn.sz][].store 消息 C 路由至 Binding Key 为 [files.cn][].\*.store 的Queue B。

### Fanout Exchange ###

路由规则

Fanout Exchange 将消息路由至所有绑定的 Queue。

使用场景

Fanout Exchange 无视 Routing Key 和 Binding Key 的匹配规则，将消息投递至所有绑定的 Queue，适用于广播消息。

示例

**图 3. Fanout Exchange[![format_png 3][]][format_png 3 1]**

如[图 3][3] 所示，Fanout Exchange 将消息投递至所有绑定的 Queue，Consumer 可以从所有 Queue 中拉取同一消息。

Producer 向 Fanout Exchange 发送消息：

消息 A 的 Routing Key 可以为任意值。

Fanout Exchange 绑定 Queue：

Fanout Exchange 和 Queue A、Queue B、Queue C 的 Binding Key 都可以为任意值。

Fanout Exchange 将消息路由至所有绑定的 Queue：

Routing Key 为任意值的消息 A 分别路由至 Binding Key 为任意值的 Queue A、Queue B、Queue C。

**参考文献**

[阿里云AMQP官网][AMQP]

[format_png]: /images/20230312/f0d7176b29144e28bde48f8e66b25135.png
[format_png 1]: http://static-aliyun-doc.oss-cn-hangzhou.aliyuncs.com/assets/img/2370232751/p66314.png
[1]: https://help.aliyun.com/document_detail/141604.html?spm=a2c4g.11186623.6.558.72246bbcViYWGT#fig-u1a-fml-6jo
[format_png 2]: /images/20230312/cf116e2706a44160af8f30f80f0de62f.png
[format_png 2 1]: http://static-aliyun-doc.oss-cn-hangzhou.aliyuncs.com/assets/img/2370232751/p66315.png
[2]: https://help.aliyun.com/document_detail/141604.html?spm=a2c4g.11186623.6.558.72246bbcViYWGT#fig-a1x-xz0-arg
[files.cn]: http://files.cn/
[files.cn.sz]: http://files.cn.sz/
[format_png 3]: /images/20230312/6584347464364bef862b7c6e8ecb047f.png
[format_png 3 1]: http://static-aliyun-doc.oss-cn-hangzhou.aliyuncs.com/assets/img/2370232751/p66316.png
[3]: https://help.aliyun.com/document_detail/141604.html?spm=a2c4g.11186623.6.558.72246bbcViYWGT#fig-qmh-4nk-v7n
[AMQP]: https://help.aliyun.com/document_detail/101631.html?spm=a2c4g.11186623.6.542.db846796RBu5HN