AI专业者开发社区 1 号成员
  •  0 回帖  •  5 浏览  •  4 个月前
   

Spring Boot整合RabbitMQ

Spring Boot 整合 RabbitMQ

目录

写在开头

最近在搭一套基于 SpringBoot 的项目，用到了 ssm+mysql+rabbitmq+redis。除了 rabbitmq 之外，其他几个都很快整合好了，唯独 rabbitmq 找了不少资料，才最终整合好，达到了预期。特此将过程记录下来，供参考。

整合流程

整合流程中的代码都为整合的关键配置及其使用。至于 SpringBoot 的基本配置，请参考Spring Boot Quick Start。

配置文件

  • pom.xml
<!-- rabbit-mq -->
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
  • application.yml
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    listener:
      simple:
        acknowledge-mode: manual # 手动应答
        concurrency: 5 # 消费端最小并发数
        max-concurrency: 10 # 消费端最大并发数
        prefetch: 5 # 一次请求中预处理的消息数量
    cache:
      channel:
        size: 50 # 缓存的 channel 数量
### 自定义配置
mq:
  defaultExchange: amqpExchange # 默认交换器
  queue: queue # 队列名
  routeKey: queue_key # 路由 key
  • MQProperties
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

@Component
@ConfigurationProperties(prefix = "mq")
public class MQProperties {
private String defaultExchange;
private String routeKey;
private String queue;


<span class="hljs-keyword">public</span> String <span class="hljs-title function_">getDefaultExchange</span><span class="hljs-params">()</span> {
    <span class="hljs-keyword">return</span> defaultExchange;
}

<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title function_">setDefaultExchange</span><span class="hljs-params">(String defaultExchange)</span> {
    <span class="hljs-built_in">this</span>.defaultExchange = defaultExchange;
}

<span class="hljs-keyword">public</span> String <span class="hljs-title function_">getRouteKey</span><span class="hljs-params">()</span> {
    <span class="hljs-keyword">return</span> routeKey;
}

<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title function_">setRouteKey</span><span class="hljs-params">(String routeKey)</span> {
    <span class="hljs-built_in">this</span>.routeKey = routeKey;
}

<span class="hljs-keyword">public</span> String <span class="hljs-title function_">getQueue</span><span class="hljs-params">()</span> {
    <span class="hljs-keyword">return</span> queue;
}

<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title function_">setQueue</span><span class="hljs-params">(String queue)</span> {
    <span class="hljs-built_in">this</span>.queue = queue;
}



}

RabbitMQ 配置

import com.switchvov.rabbitmq.constant.MQProperties;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@EnableRabbit
public class RabbitMQConfig {
@Autowired
private MQProperties mqProperties;


<span class="hljs-meta">@Bean</span>
<span class="hljs-keyword">public</span> Queue <span class="hljs-title function_">queue</span><span class="hljs-params">()</span> {
    <span class="hljs-type">boolean</span> <span class="hljs-variable">durable</span> <span class="hljs-operator">=</span> <span class="hljs-literal">true</span>;
    <span class="hljs-type">boolean</span> <span class="hljs-variable">exclusive</span> <span class="hljs-operator">=</span> <span class="hljs-literal">false</span>;
    <span class="hljs-type">boolean</span> <span class="hljs-variable">autoDelete</span> <span class="hljs-operator">=</span> <span class="hljs-literal">false</span>;
    <span class="hljs-keyword">return</span> <span class="hljs-keyword">new</span> <span class="hljs-title class_">Queue</span>(mqProperties.getQueue(), durable, exclusive, autoDelete);
}

<span class="hljs-meta">@Bean</span>
<span class="hljs-keyword">public</span> DirectExchange <span class="hljs-title function_">defaultExchange</span><span class="hljs-params">()</span> {
    <span class="hljs-type">boolean</span> <span class="hljs-variable">durable</span> <span class="hljs-operator">=</span> <span class="hljs-literal">true</span>;
    <span class="hljs-type">boolean</span> <span class="hljs-variable">autoDelete</span> <span class="hljs-operator">=</span> <span class="hljs-literal">false</span>;
    <span class="hljs-keyword">return</span> <span class="hljs-keyword">new</span> <span class="hljs-title class_">DirectExchange</span>(mqProperties.getDefaultExchange(), durable, autoDelete);
}

<span class="hljs-meta">@Bean</span>
<span class="hljs-keyword">public</span> Binding <span class="hljs-title function_">binding</span><span class="hljs-params">()</span> {
    <span class="hljs-keyword">return</span> BindingBuilder.bind(queue())
            .to(defaultExchange())
            .with(mqProperties.getRouteKey());
}



}

RabbitMQ 生产者

import com.switchvov.rabbitmq.constant.MQProperties;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMQTest {
@Autowired
private RabbitTemplate rabbitTemplate;


<span class="hljs-meta">@Autowired</span>
<span class="hljs-keyword">private</span> MQProperties mqProperties;

<span class="hljs-meta">@Test</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title function_">testSendMessage</span><span class="hljs-params">()</span> {
    rabbitTemplate.convertAndSend(mqProperties.getDefaultExchange(), 
        mqProperties.getRouteKey(), <span class="hljs-string">"发送了一条信息"</span>);
}



}

RabbitMQ 消费者

import com.switchvov.rabbitmq.common.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;

@Service
public class RabbitMQService {
private static final Logger LOGGER = LoggerFactory.getLogger(RabbitMQService.class);


<span class="hljs-meta">@RabbitListener(queues = "${mq.queue}")</span>
<span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title function_">receive</span><span class="hljs-params">(String payload, Channel channel,
                    <span class="hljs-meta">@Header(AmqpHeaders.DELIVERY_TAG)</span> <span class="hljs-type">long</span> tag)</span> {
    LOGGER.info(<span class="hljs-string">"消费内容为：{}"</span>, payload);
    RabbitMQUtils.askMessage(channel, tag, LOGGER);
}



}

手动应答简单工具类

import com.rabbitmq.client.Channel;
import org.slf4j.Logger;

import java.io.IOException;


public final class RabbitMQUtils {


<span class="hljs-keyword">public</span> <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> <span class="hljs-title function_">askMessage</span><span class="hljs-params">(Channel channel, <span class="hljs-type">long</span> tag, <span class="hljs-keyword">final</span> Logger logger)</span> {
    askMessage(channel, tag, logger, <span class="hljs-literal">false</span>);
}

<span class="hljs-keyword">public</span> <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> <span class="hljs-title function_">askMessage</span><span class="hljs-params">(Channel channel, <span class="hljs-type">long</span> tag, <span class="hljs-keyword">final</span> Logger logger, <span class="hljs-type">boolean</span> multiple)</span> {
    <span class="hljs-keyword">try</span> {
        channel.basicAck(tag, multiple);
    } <span class="hljs-keyword">catch</span> (IOException e) {
        logger.error(<span class="hljs-string">"RabbitMQ，IO异常，异常原因为：{}"</span>, e.getMessage());
    }
}

<span class="hljs-keyword">public</span> <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> <span class="hljs-title function_">rejectMessage</span><span class="hljs-params">(Channel channel, <span class="hljs-type">long</span> tag, <span class="hljs-keyword">final</span> Logger logger)</span> {
    rejectMessage(channel, tag, logger, <span class="hljs-literal">false</span>, <span class="hljs-literal">false</span>);
}

<span class="hljs-keyword">public</span> <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> <span class="hljs-title function_">rejectAndBackMQ</span><span class="hljs-params">(Channel channel, <span class="hljs-type">long</span> tag, <span class="hljs-keyword">final</span> Logger logger)</span> {
    rejectMessage(channel, tag, logger, <span class="hljs-literal">false</span>, <span class="hljs-literal">true</span>);
}

<span class="hljs-keyword">public</span> <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> <span class="hljs-title function_">rejectMessage</span><span class="hljs-params">(Channel channel, <span class="hljs-type">long</span> tag, <span class="hljs-keyword">final</span> Logger logger, <span class="hljs-type">boolean</span> multiple, <span class="hljs-type">boolean</span> request)</span> {
    <span class="hljs-keyword">try</span> {
        channel.basicNack(tag, multiple, request);
    } <span class="hljs-keyword">catch</span> (IOException e) {
        logger.error(<span class="hljs-string">"RabbitMQ，IO异常，异常原因为：{}"</span>, e.getMessage());
    }
}



}

参考文档

  1. RabbitMQ 消息队列（一）: Detailed Introduction 详细介绍
    • 简单的了解下 RabbitMQ 的系统架构和各种术语
  2. Spring Boot 中使用 RabbitMQ
    • 简单的 Spring Boot 和 RabbitMQ 整合资料
  3. Spring boot 集成 RabbitMQ
    • 弄清楚了自定义配置queue、exchange、binding
  4. Spring RabbitMQ - using manual channel acknowledgement on a service with @RabbitListener configuration
    • 接收者部分基本上就是按照这上面的回答实现的
  5. spring boot / cloud (十九) 并发消费消息, 如何保证入库的数据是最新的?
    • 并发消费配置

分享并记录所学所见