Monodyee

这个屌丝很懒，什么也没留下！

热门标签

RabbitMq + Nacos整合_rabbitmq nacos

作者：Monodyee | 2024-04-13 05:56:43

踩

rabbitmq nacos

RabbitMq + Nacos入门整合

创建父工程:rabbitmqnacos2021

1.选择Maven项目，然后选择maven-archetype-site，最后点击Next

2.填好自己的GAV，Next

3.选好自己的Maven配置

选择完Maven依赖后，一直Next直至Finish即可.

4.字符编码配置

5.注解生效激活

6.修改JAVA编译版本为8

7.修改父工程的POM文件

父工程POM依赖

 <?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>com.rabbitmq.nacos</groupId>
  <artifactId>rabbitmq-nacos-2021</artifactId>
  <version>1.0-SNAPSHOT</version>

  <!--统一管理jar包版本-->
  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <spring.boot.version>2.2.2.RELEASE</spring.boot.version>
    <spring.cloud.alibaba.version>2.1.0.RELEASE</spring.cloud.alibaba.version>
    <spring.amqp.rabbit.version>2.2.2.RELEASE</spring.amqp.rabbit.version>
  </properties>
  
  <dependencyManagement>

    <dependencies>
      <!--spring cloud alibaba 2.1.0.RELEASE-->
      <dependency>
        <groupId>com.alibaba.cloud</groupId>
        <artifactId>spring-cloud-alibaba-dependencies</artifactId>
        <version>${spring.cloud.alibaba.version}</version>
        <type>pom</type>
        <scope>import</scope>
      </dependency>

      <!--amqp + rabbitMq 2.2.2.RELEASE-->
      <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
        <version>${spring.amqp.rabbit.version}</version>
      </dependency>

      <!--springboot 2.2.2.RELEASE-->
      <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-dependencies</artifactId>
        <version>${spring.boot.version}</version>
        <type>pom</type>
        <scope>import</scope>
      </dependency>

    </dependencies>
  </dependencyManagement>

  <!--第三方maven私服-->
  <repositories>
    <repository>
      <id>nexus-aliyun</id>
      <name>Nexus aliyun</name>
      <url>http://maven.aliyun.com/nexus/content/groups/public</url>
      <releases>
        <enabled>true</enabled>
      </releases>
      <snapshots>
        <enabled>false</enabled>
      </snapshots>
    </repository>
  </repositories>

</project>
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

至此，父工程的搭建已经结束。下面开始其他模块的搭建工作。

创建消费者子工程:cloud-rabbit-consumer

1.在父工程名上右键后选择如下图红框所示:

2.和父工程不同的是，此处要选择maven-archetype-quickstart(maven jar项目)，不能再选择site聚合工程.

3.其余步骤请参考父工程创建步骤的2、3步。

4.修改子工程的POM文件

子工程:cloud-rabbit-consumer POM依赖

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>rabbitmq-nacos-2021</artifactId>
        <groupId>com.rabbitmq.nacos</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>cloud-rabbit-consumer</artifactId>

    <dependencies>
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
        </dependency>

        <!--web + actuator-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <!--amqp + rabbitMq-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
    </dependencies>

</project>

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

5.写yml配置

# 本服务端口
server:
  port: 56723

# 本服务应用名称
spring:
  application:
    name: cloud-rabbit-consumer

# Nacos配置地址
  cloud:
    nacos:
      discovery:
        server-addr: localhost:8848

#RabbitMq配置
  rabbitmq:
    username: admin
    password: admin
    virtual-host: /
    host: 此处写你的RabbitMq服务地址
    port: 此处写你的RabbitMq端口
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

6.编写服务主启动类

package com.rabbitmq.nacos;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;

@SpringBootApplication
@EnableDiscoveryClient
public class ConsumerMain56723 {
    public static void main(String[] args) {
        SpringApplication.run(ConsumerMain56723.class, args);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

7.编写RabbitMq交换机绑定配置类

package com.rabbitmq.nacos.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 关于配置类的创建建议放在消费者这边，因为消费者启动的时候如果MQ中的对应队列没有创建，则会出现异常。
 * 而生产者启动的时候没有相关的交换机和队列也不会出现异常，
 * 只是在往对应的交换机中发送消息的时候会有Error信息说找不到该交换机。
 */
@Configuration
public class RabbitMqBindConfig {

    /**
     * 声明交换机
     */
    @Bean
    public FanoutExchange fanoutExchange(){
        //参数说明:
        //name:交换机的名称
        //durable:true、false 服务器重启是否会保留下来Exchange。警告：仅设置此选项，不代表消息持久化。即不保证重启后消息还在。原文：true if we are declaring a durable exchange (the exchange will survive a server restart)
        //autoDelete:true、false. 当已经没有消费者时，服务器是否可以删除该Exchange。原文1：true if the server should delete the exchange when it is no longer in use。
        return new FanoutExchange("rabbit-fanout-exchange",true,false);
    }


    /**
     * 声明队列
     */
    @Bean
    public Queue emailQueue(){
        //参数说明:
        //name:交换机的名称
        //durable:true、false 服务器重启是否会保留下来Exchange。警告：仅设置此选项，不代表消息持久化。即不保证重启后消息还在。原文：true if we are declaring a durable exchange (the exchange will survive a server restart)
        return new Queue("email.queue",true);
    }

    /**
     * 绑定交换机与队列
     */
    @Bean
    public Binding emailBinding(){
        //该bind方法意思是，将某队列绑定(bind)到(to)某交换机上
        return BindingBuilder.bind(emailQueue()).to(fanoutExchange());
    }

}

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

8.编写消费者Service，消费端的监听方式有三种，基于配置类的有两种，不需要配置类的一种:如下。

8.1 基于配置类，使用@RabbitListener注解的方式进行监听

package com.rabbitmq.nacos.consumer;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class ConsumerConfigListener {

    //queues中填写想要监听的队列名称，注意，请勿写错，否则无法获取消息

    /**
     * 说明:监听队列:email.queue的消息
     * @param message RabbitMQ的Meaasge，又称消息，是服务器与应用程序之间传递的数据，由Properties和Body组成， Properties可以对消息进行修饰，如消息的优先级、传输格式（如JSON）、延迟等高级特性，Body则就是消息体内容；
     * @param channel Channel是我们与RabbitMQ打交道的最重要的一个接口，我们大部分的业务操作是在Channel这个接口中完成的，
     *                包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。
     *                如果每一次访问RabbitMQ都建立一个Connection，在消息量大的时候建立TCP Connection的开销将是巨大的，效率也较低。
     *                Channel是在connection内部建立的逻辑连接，如果应用程序支持多线程，
     *                通常每个thread创建单独的channel进行通讯，AMQP method包含了channel id帮助客户端和message broker识别channel，
     *                所以channel之间是完全隔离的。Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销
     *                
     * 注意: channel的说明引用了博主「小小角色熊」的原创文章 原文链接：https://blog.csdn.net/xiaobaixiongxiong/article/details/89071917
     *                
     *   channel常见的一个作用是可以手动ack,具体使用因技术受限就不做过多说明以免产生误导，还请大家自行查阅资料。             
     */
    @RabbitListener(queues = "email.queue")
    public void smsConsumerListener(Message message, Channel channel) {

        String msg = new String(message.getBody());
        
		System.out.println("email.queue-------------->" + msg);
    }
}
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

8.2 基于配置类，使用@RabbitListener与@RabbitHandler注解的方式进行监听

package com.rabbitmq.nacos.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = "email.queue")
public class ConsumerConfigListenerTwo {

    /**
     * @param mes 注意，此参数需要根据监听的数据类型进行定义，若数据类型为String则使用String监听，否则报错。
     */
    @RabbitHandler
    public void smsConsumer(String msg) {
        System.out.println("email.queue-------------->" + msg);
    }
}

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

8.2 舍弃配置类，使用@RabbitListener与@RabbitHandler注解的方式进行监听,
之所以不需要配置类，是因为在@RabbitListener就已经使用注解的属性进行了队列的声明和绑定。

package com.rabbitmq.nacos.consumer;


import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

// bindings其实就是用来确定队列和交换机绑定关系
@RabbitListener(bindings =@QueueBinding(
        // email.queue 是队列名字，这个名字你可以自定随便定义。
        value = @Queue(value = "email.queue",autoDelete = "false"),
        // rabbit-fanout-exchange 是交换机的名字 必须和生产者保持一致
        exchange = @Exchange(value = "rabbit-fanout-exchange",
                // 这里是确定的rabbitmq模式是：fanout 是以广播模式 、 发布订阅模式
                // 还有Topic与Direct模式
                type = ExchangeTypes.FANOUT)))
@Component
public class ConsumerConfigListenerThree {
    /**
     * @RabbitHandler 代表此方法是一个消息接收的方法。
     * @param mes 注意，此参数需要根据监听的数据类型进行定义，若数据类型为String则使用String监听，否则报错。
     */
    @RabbitHandler
    public void smsConsumer(String msg) {
        System.out.println("email.queue-------------->" + msg);
    }
}
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

方法三的验证结果如下图，可明显看到控制台无异常，RabbitMq的管理界面也新建了对应的交换机与队列:

消费者搭建完毕后的工程结构请看图:

以上三种监听的方式，根据业务场景自行选择。

至此，消费者服务搭建完毕，下面开始搭建生产者。

创建生产者子工程:cloud-rabbit-producer

1.子工程创建步骤请参考消费者的步骤，此处不再赘述。

2.POM依赖与消费者一致，注意，请勿Ctrl-a全部复制POM文件到生产者，因为artifactId不同。不要太懒，多动手多动脑总是好的，复制所需依赖就够啦。

3.写yml配置,与消费者服务不同的是，只改了本服务的端口号与应用名称.

# 本服务端口
server:
  port: 56722

# 本服务应用名称
spring:
  application:
    name: cloud-rabbit-producer

# Nacos配置地址
  cloud:
    nacos:
      discovery:
        server-addr: localhost:8848

#RabbitMq配置
  rabbitmq:
    username: admin
    password: admin
    virtual-host: /
    host: 此处写你的RabbitMq服务地址
    port: 此处写你的RabbitMq端口
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

4.生产者主启动类

package com.rabbitmq.nacos;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;

/**
 * Hello world!
 */
@SpringBootApplication
@EnableDiscoveryClient
public class ProducerMain56722 {
    public static void main(String[] args) {
        SpringApplication.run(ProducerMain56722.class, args);
    }
}

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

5.生产者消息生产Service

package com.rabbitmq.nacos.producer;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.UUID;

/**
 * 消息生产者
 */
@Service
public class ProducerMsgService {
    @Resource
    private RabbitTemplate rabbitTemplate;

    public void productToPushEmailQueue() {
        // 模拟发送邮件
        String emailmsg = UUID.randomUUID().toString();
 		System.out.println("发送消息:" + emailmsg);
        rabbitTemplate.convertAndSend( "rabbit-fanout-exchange", "", emailmsg);
    }
}

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

6.生产者消息生产Controller

package com.rabbitmq.nacos.controller;

import com.rabbitmq.nacos.producer.ProducerMsgService;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;

/**
 * 出发消息生产入口，当然也可以选择测试类
 */
@RestController
public class ProducerMsgController {

    @Resource
    private ProducerMsgService producerMsgService;
    
    @GetMapping("/mq/producer")
    public void rabbitMqProducer() {
        producerMsgService.productToPushEmailQueue();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21