当前位置:   article > 正文

RabbitMq + Nacos整合_rabbitmq nacos

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

生产者最后的工程结构请看图:

7.通过PostMan发送请求后成功推送消息:

RabbitMq中也有对应的消息

下面来看消费情况,可以看到启动后成功在控制台打印了队列中的消息:

nacos中服务情况:

服务:rabbitmqnacos2021的整体结构如图:

以上就是RabbitMq + Nacos整合的搭建过程。

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/Monodyee/article/detail/414769
推荐阅读
相关标签
  

闽ICP备14008679号