RabbitMQ学习笔记 - RabbitMQ整合SpringCloud

RabbitMQ整合SpringCloud

    你好！欢迎来到Java成长笔记，主要是用于相互交流，相互学习，也希望分享能帮到大家，如有错误之处，希望指正，谢谢！

整合RabbitMQ简介

1、SpringCloudStream定位是兼容主流消息中间件的集成使用，减少不同消息中间件集成的配置，它整合RabbitMQ配置相较简单，不需要定义相应的交换机、队列、以及关系绑定，使相应的配置减少。
2、SpringCloudStream通过上层结构上的处理，使消息生产端、消费端可以多样化，不需要拘泥于生产消费端使用相同的消息中间件。例如：生产端可以使用RabbitMQ，而消费端可以使用Kafka，让开发者省去了相应不同的配置的集成，开发者只需要使用好相应的几个注解，就能实现高性能的生产和消费的场景。
3、当然它也有一个非常大的问题就是不能实现消息的可靠性投递，也就是不能保证消息的100%可靠性，会存在少量的消息丢失。

引入依赖配置

主要依赖

// 指定统一配置
<properties>
    <java.version>1.8</java.version>
    <spring.cloud-version>Greenwich.SR6</spring.cloud-version>
    <spring.boot-version>Brussels-SR17</spring.boot-version>
</properties>

// springcloud依赖
<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>${spring.cloud-version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

// 其它配置
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    </dependency>
</dependencies>

使用注解说明

@Output：输出注解，用于定义发送消息接口
@input：输入注解，用于定义消息的消费者接口
@StreamListener：用于定义监听方法的注解
@EnableBinding：启动绑定关系

消息生产端

生产端配置说明

spring:
  cloud:
    stream:
      binders:
        defaultRabbit:                          # 定义的名称，用于bidding整合
          type: rabbit                          # 指定消息类型
          environment:
            spring:
              rabbitmq:                         # rabbitmq 配置信息
                addresses: 127.0.0.1:5672       # rabbitmq 连接地址
                username: rabbitmq              # rabbitmq 用户
                password: 123456                # rabbitmq 连接密码
                virtual-host: /                 # 虚拟路径
      bindings:
        userOutPutChannel:
          destination: exchange_cloud           # Exchange名称，交换模式默认topic，把stream输出通道绑定到exchange_cloud交换机
          group: userGroup                      # 分组名称，生产端、消费端名称需要一致
          default-binder: defaultRabbit         # 和上面定义的 binders:defaultRabbit需要一致
          content-type: application/json        # 设置消息类型 为json

生产端代码

展示如下：

// 定义输出类型的通道
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.stereotype.Component;

@Component
public interface Barista {

    final static String OUTPUT_CHANNEL = "userOutPutChannel";

    /*
     * @Description: 定义一个输出类型的通道
     * @Author ly
     * @param []
     * @return org.springframework.messaging.MessageChannel
     * @date 2021/3/22 17:07
     */
    @Output(Barista.OUTPUT_CHANNEL)
    MessageChannel userOutPutChannel();

}

// 实现类封装
import com.alibaba.fastjson.JSON;
import com.show.service.Barista;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Map;

@EnableBinding(Barista.class)
@Service
@Slf4j
public class RabbitSender {

    @Resource
    @Output(Barista.OUTPUT_CHANNEL)
    private MessageChannel channel;

    @Resource
    private Barista barista;

    /*
     * @Description: channel 发送消息
     * @Author ly
     * @date 2021/3/23 11:12
     */
    public void sendMsg (String msg) {
        channel.send(MessageBuilder.withPayload(msg).build());
        log.error("channel消息发送成功：{} " + msg);
    }

    /*
     * @Description:  barista 发送消息
     * @Author ly
     * @date 2021/3/23 11:12
     */
   public void sendMessage (Object message, Map<String, Object> properties) {
        final MessageHeaders messageHeaders = new MessageHeaders(properties);
        final Message msg = MessageBuilder.createMessage(message, messageHeaders);
        final boolean sendResult = barista.userOutPutChannel().send(msg);
        log.error("barista消息发送成功:{},sendResult:{} " + JSON.toJSONString(msg), sendResult);
    }

}

// 测试类处理
import com.google.common.collect.ImmutableMap;
import com.show.model.User;
import com.show.service.impl.RabbitSender;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.test.context.junit4.SpringRunner;
import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.Map;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = ECloudProducerApplicationTests.class)
@ComponentScan(basePackages = {"com.show.*"})
@Slf4j
public class ECloudProducerApplicationTests {

    @Resource
    private RabbitSender rabbitSender;

    @Test
    public void sendMessage () {
        final String message = "Hello RabbitMQ";
        rabbitSender.sendMsg(message);
    }

    @Test
    public void sendRabbitMessage () {
        final Map<String, Object> properties = ImmutableMap.of("cloud-stream", "cloud-stream");
        final User user = new User("simon","123456", 22,  new BigDecimal(100));
        rabbitSender.sendMessage(user, properties);
    }

}

消费端代码

消费端配置说明

spring:
  cloud:
    stream:
      binders:
        defaultRabbit:                              # 此配置为rabbitmq配置说明
          type: rabbit
          environment:
            spring:
              rabbitmq:
                addresses: 127.0.0.1:5672
                username: rabbit
                password: 123456
                virtual-host: /
      bindings:
        userInChannel:                              # 定义输入管道名称          
          destination: exchange_cloud               # 交换模式默认是topic，把stream的消息输出通道绑定到exchange_cloud交换器
          group: userGroup                          # 分组名称与生产端名称一致
          content-type: application/json            # 消费端消息类型 json
          default-binder: defaultRabbit             # 与binders:defaultRabbit名称一致 环境名称
          consumer:
            concurrency: 1                          # 默认监听数量
      rabbit:
        bindings:
          userInChannel:                            # 和bindings:userInChannel 名称一致
            consumer:                               # 消费端配置     
              requeue-reject: false                 # 是否支持return
              acknowledge-mode: MANUAL              # 签收模式 手动签收
              recovery-interval: 3000               # 3s重新连接 
              durable-subscription: true            # 是否启动持久化订阅
              max-concurrency: 5                    # 最大监听数量

消费端代码

// 定义一个输入通道
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.stereotype.Component;

@Component
public interface Barista {
      
    String INPUT_CHANNEL = "userInChannel";

    /*
     * @Description: 定义一个输入类型的通道
     * @Author ly
     * @param []
     * @return org.springframework.messaging.SubscribableChannel
     * @date 2021/3/22 17:34
     */
    @Input(Barista.INPUT_CHANNEL)
    SubscribableChannel userInChannel();

}  

// 定义输入通道监听类
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.show.service.Barista;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;

@EnableBinding(Barista.class)
@Service
@Slf4j
public class MQReceiver {

    @StreamListener(Barista.INPUT_CHANNEL)
    public void receiver(Message message) throws Exception {
        Channel channel = (Channel) message.getHeaders().get(AmqpHeaders.CHANNEL);
        Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        log.error("消费完毕：{}, Object：{}", System.currentTimeMillis(), JSON.toJSON(message));
        channel.basicAck(deliveryTag, false);
    }

}

接收消息返回信息示例

sendMessage测试返回

对象消息返回信息示例
sendRabbitMessage测试返回

本章完结，后续还会持续更新，分享Java成长笔记，希望我们能一起成长。如果你觉得我的分享有用，记得点赞和关注哦！这对我是最好的鼓励。谢谢！

PS：转载请注明出处！

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容