Springboot + netty + rabbitmq + myBatis

news/2024/10/4 15:16:04/

目录

  • 0.为什么用消息队列
  • 1.代码文件创建结构
  • 2.pom.xml文件
  • 3.三个配置文件开发和生产环境
  • 4.Rabbitmq 基础配置类 TtlQueueConfig
  • 5.建立netty服务器 + rabbitmq消息生产者
  • 6.建立常规队列的消费者 Consumer
  • 7.建立死信队列的消费者 DeadLetterConsumer
  • 8.建立mapper.xml文件
  • 9.建立mapper文件接口
  • 10.建立接口ProducerController 测试
  • 11.测试接口请求1
  • 12.测试接口请求2
  • 13.网络助手测试NetAssist.exe
  • 14.观察rabbitmq界面管理简单介绍

0.为什么用消息队列

  1. 流量消峰
  2. 应用解耦
  3. 异步确认

1.代码文件创建结构

在这里插入图片描述

2.pom.xml文件

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.3.4.RELEASE</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>com.test</groupId><artifactId>demo</artifactId><version>0.0.1-SNAPSHOT</version><name>demo</name><description>Demo project for Spring Boot</description><packaging>war</packaging><url/><licenses><license/></licenses><developers><developer/></developers><scm><connection/><developerConnection/><tag/><url/></scm><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId></dependency><dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.86.Final</version> <!-- 根据需要选择版本 --></dependency><dependency><groupId>com.baomidou</groupId><artifactId>dynamic-datasource-spring-boot-starter</artifactId><version>3.5.0</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>easyexcel</artifactId><version>3.1.1</version></dependency><dependency><groupId>org.mybatis.spring.boot</groupId><artifactId>mybatis-spring-boot-starter</artifactId><version>2.2.0</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><scope>runtime</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><configuration><excludes><exclude><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></exclude></excludes></configuration></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-surefire-plugin</artifactId><configuration><skip>true</skip></configuration></plugin></plugins></build>
</project>

3.三个配置文件开发和生产环境

在这里插入图片描述
文件一 application.properties

#开发环境
spring.profiles.active=dev
#生产环境
#spring.profiles.active=prod

文件二 application-dev.properties 开发环境

spring.application.name=demo
server.servlet.context-path=/demo1
server.port=1001
#spring.main.allow-circular-references=true
spring.rabbitmq.host=实际ip地址
spring.rabbitmq.port=5672
spring.rabbitmq.username=root
spring.rabbitmq.password=root
spring.rabbitmq.virtual-host=/
##创建单线程监听容器 本项目目前用的是单线程  这里预期值2000 需根据实际情况调整
spring.rabbitmq.listener.simple.prefetch=2000
##创建多线程监听容器
#spring.rabbitmq.listener.direct.prefetch=2000
#spring.rabbitmq.listener.simple.acknowledge-mode=auto
# application.properties 示例
spring.rabbitmq.listener.simple.acknowledge-mode=MANUAL
#开启消息确认机制
spring.rabbitmq.publisher-confirm-type=correlated
spring.rabbitmq.publisher-returns=truenetty.server.port=1002
netty.server.bossThreads=1
netty.server.workerThreads=1server.max-http-header-size=655360
mybatis.mapper-locations=classpath:mapper/*Mapper.xml
mybatis.type-aliases-package=com.mt.entity
spring.mvc.pathmatch.matching-strategy=ant_path_matcher
logging.level.com.ysd.mapper=info
logging.file.name=demo.log
logging.level.com.mt.mapper=info
#mybatis.configuration.log-impl=org.apache.ibatis.logging.stdout.StdOutImpl
mybatis.configuration.map-underscore-to-camel-case=true
pagehelper.helper-dialect=mysql
pagehelper.reasonable=true
spring.main.allow-circular-references=true
spring.jackson.default-property-inclusion=non_null#<!--=====================数据库1 ====================-->
spring.datasource.dynamic.英文数据库名称1.url=jdbc:mysql://ip地址:3306/英文数据库名称?serverTimezone=UTC&useUnicode=true&characterEncoding=utf8&useSSL=false
spring.datasource.dynamic.datasource.英文数据库名称1.username=root
spring.datasource.dynamic.datasource.英文数据库名称1.password=root
spring.datasource.dynamic.datasource.英文数据库名称1.driver-class-name=com.mysql.cj.jdbc.Driver
#<!--=====================数据库2 ====================-->
spring.datasource.dynamic.英文数据库名称2.url=jdbc:mysql://ip地址:3306/英文数据库名称?serverTimezone=UTC&useUnicode=true&characterEncoding=utf8&useSSL=false
spring.datasource.dynamic.datasource.英文数据库名称2.username=root
spring.datasource.dynamic.datasource.英文数据库名称2.password=root
spring.datasource.dynamic.datasource.英文数据库名称2.driver-class-name=com.mysql.cj.jdbc.Driver

4.Rabbitmq 基础配置类 TtlQueueConfig

rabbitmq基础信息配置已经在application-dev.properities中进行配置过一部分

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;@Configuration
public class TtlQueueConfig {public static final String X_EXCHANGE = "X";public static final String QUEUE_A = "QA";public static final String QUEUE_B = "QB";public static final String Y_DEAD_LETTER_EXCHANGE = "Y";public static final String DEAD_LETTER_QUEUE = "QD";@Bean("xExchange")public DirectExchange xExchange() {return new DirectExchange(X_EXCHANGE);}@Bean("yExchange")public DirectExchange yExchange() {return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);}@Bean("queueA")public Queue queueA() {Map<String, Object> args = new HashMap<>(3);//声明当前队列绑定的死信交换机args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);//声明当前队列的死信路由 keyargs.put("x-dead-letter-routing-key", "YD");//声明队列的 TTLargs.put("x-message-ttl", 150000);//超出150秒没有被消费  就会进入死信队列return QueueBuilder.durable(QUEUE_A).withArguments(args).build();}@Beanpublic Binding queueaBindingX(@Qualifier("queueA") org.springframework.amqp.core.Queue queueA,@Qualifier("xExchange") DirectExchange xExchange) {return BindingBuilder.bind(queueA).to(xExchange).with("XA");}//声明队列 B ttl 为 40s 并绑定到对应的死信交换机@Bean("queueB")public Queue queueB() {Map<String, Object> args = new HashMap<>(3);//声明当前队列绑定的死信交换机args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);//声明当前队列的死信路由 keyargs.put("x-dead-letter-routing-key", "YD");//声明队列的 TTLargs.put("x-message-ttl", 40000);//超出40秒没有被消费  就会进入死信队列return QueueBuilder.durable(QUEUE_B).withArguments(args).build();}//声明队列 B 绑定 X 交换机@Beanpublic Binding queuebBindingX(@Qualifier("queueB") Queue queue1B,@Qualifier("xExchange") DirectExchange xExchange) {return BindingBuilder.bind(queue1B).to(xExchange).with("XB");}//声明死信队列 QD@Bean("queueD")public Queue queueD() {return new Queue(DEAD_LETTER_QUEUE);}//声明死信队列 QD 绑定关系@Beanpublic Binding deadLetterBindingQAD(@Qualifier("queueD") Queue queueD,@Qualifier("yExchange") DirectExchange yExchange) {return BindingBuilder.bind(queueD).to(yExchange).with("YD");}@Beanpublic SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(ConnectionFactory connectionFactory) {SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory = new SimpleRabbitListenerContainerFactory();//这个connectionFactory就是我们自己配置的连接工厂直接注入进来simpleRabbitListenerContainerFactory.setConnectionFactory(connectionFactory);//这边设置消息确认方式由自动确认变为手动确认simpleRabbitListenerContainerFactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);//设置消息预取数量// simpleRabbitListenerContainerFactory.setPrefetchCount(1);return simpleRabbitListenerContainerFactory;}
}

netty__rabbitmq_278">5.建立netty服务器 + rabbitmq消息生产者

创建服务器类 初始化启动服务器NettyServer

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;@Component
@Slf4j
public class NettyServer implements ApplicationRunner {@Autowiredprivate RabbitTemplate rabbitTemplate;//有可能在项目初始化的时候加载不出来导致项目隐形报错//加载application-dev.proerities文件中 对应参数配置项@Value("${netty.server.port}")private int port;public int getPort () {return port;}@Overridepublic void run(ApplicationArguments args) throws Exception {log.info("netty服务启动端口"+getPort());EventLoopGroup bossGroup = new NioEventLoopGroup(); // 用于接收进来的连接EventLoopGroup workerGroup = new NioEventLoopGroup(); // 用于处理已经被接收的连接ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class) // 使用Nio的通道类型.childHandler(new ChannelInitializer<SocketChannel>() { // 添加一个处理器来处理接收到的数据@Overridepublic void initChannel(SocketChannel ch) throws Exception {ChannelPipeline p = ch.pipeline();p.addLast(new StringDecoder());p.addLast(new StringEncoder());p.addLast(new SimpleChannelInboundHandler<String>() {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {//netty监听端口消息发送消息通过rabbitmq生产者发送到消息队列sendMessage(ctx,msg);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();}});}});// 绑定端口，开始接收进来的连接ChannelFuture f = b.bind(getPort()).sync();// 等待服务器socket关闭f.channel().closeFuture().sync();}//netty监听端口消息发送消息通过rabbitmq生产者发送到消息队列中//充当消息的生产者发送public  void  sendMessage(ChannelHandlerContext ctx, String msg){//  接收netty监听信息来源作为消息生产者rabbitTemplate.convertAndSend("X","XA","来自QA"+msg);ctx.writeAndFlush("===>"+msg);}
}

6.建立常规队列的消费者 Consumer

import com.rabbitmq.client.Channel;
import com.test.demo.service.MessageProcessService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;@Slf4j
@Component
public class Consumer {@Resourceprivate MessageProcessService messProceService;private Map<Integer, Message> messageMap = new HashMap<>();private  int i =1;private  int j =1;@RabbitListener(queues = "QA")public void receiveQA(Message message, Channel channel) throws InterruptedException, IOException {i++;synchronized (this) {messageMap.put(i,message);if (messageMap.size() >= 1500) {// 模拟数据库插入操作System.out.println("模拟插入数据库操作，QA队列处理消息数：" + messageMap.size());processMessagesBatch(channel);System.out.println("模拟插入成功, 准备进行下一次收集");}}}//    @RabbitListener(queues = "QB")
//    public void receiveQB(Message message, Channel channel) throws IOException {
//        String msg = new String(message.getBody());
//        log.info("当前时间：{},监听QB队列信息{}", new Date().toString(), msg);
//    }public static class SleepUtils {public static void sleep(int second) {try {Thread.sleep(1000 * second);} catch (InterruptedException _ignored) {Thread.currentThread().interrupt();}}}private void processMessagesBatch(Channel channel) throws  IOException {List<String> list = new ArrayList<>();long startTime = System.currentTimeMillis();Map<Integer, Message> tempMessageMap = new HashMap<>(messageMap);messageMap.clear();for (Map.Entry<Integer, Message> map : tempMessageMap.entrySet()) {list.add("C" + (j++) + ":===>" + "QA队列收集" + new String(map.getValue().getBody()));}int count = messProceService.add(list);if (count == list.size()) {System.out.println("插入数据成功");for (Map.Entry<Integer, Message> map : tempMessageMap.entrySet()) {channel.basicAck(map.getValue().getMessageProperties().getDeliveryTag(), false);}list.clear();}long durationSeconds = (System.currentTimeMillis() - startTime) / 1000;System.out.println("插入1500条数据执行时间: " + durationSeconds);}
}

7.建立死信队列的消费者 DeadLetterConsumer

import com.rabbitmq.client.Channel;
import com.test.demo.service.MessageProcessService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;@Slf4j
@Component
public class DeadLetterConsumer {@Resourceprivate MessageProcessService messProceService;private Map<Integer, Message> messageMap = new HashMap<>();private int i = 1;private  int j =1;@RabbitListener(queues = "QD")public void receiveD(Message message, Channel channel) throws IOException {i++;synchronized (this) {messageMap.put(i, message);if (messageMap.size() >= 100) {// 模拟数据库插入操作System.out.println("模拟插入数据库操作，死信队列处理消息数：" + messageMap.size());processMessagesBatch(channel);System.out.println("模拟插入成功, 准备进行下一次收集");// 确认当前消息，以便RabbitMQ知道它可以释放此消息}}}private void processMessagesBatch(Channel channel) throws  IOException {List<String> list = new ArrayList<>();// 复制当前消息映射以避免在迭代时修改Map<Integer, Message> tempMessageMap = new HashMap<>(messageMap);messageMap.clear(); long startTime = System.currentTimeMillis();for (Map.Entry<Integer, Message> map : tempMessageMap.entrySet()) {list.add("C" + (j++) + ":===>" + "死信队列收集" + new String(map.getValue().getBody()));}int count = messProceService.add(list);long durationSeconds = (System.currentTimeMillis() - startTime) / 1000;System.out.println("插入50条数据执行时间: " + durationSeconds);if (count == list.size()) {System.out.println("插入数据成功");list.clear();for (Map.Entry<Integer, Message> map : tempMessageMap.entrySet()) {channel.basicAck(map.getValue().getMessageProperties().getDeliveryTag(), false);}}}}

8.建立mapper.xml文件

MessageProcessMapper.xml

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.test.demo.mapper.MessageProcessMapper"><insert id="add">INSERT INTO t_xinyang_direct (direct,create_date)VALUES<foreach collection="list" item="item" separator=",">(#{item},SYSDATE())</foreach></insert>
</mapper>

9.建立mapper文件接口

MessageProcessMapper


import com.baomidou.dynamic.datasource.annotation.DS;
import org.apache.ibatis.annotations.Mapper;import java.util.List;@Mapper
@DS("英文数据库名称1")
public interface MessageProcessMapper {int add(List<String> list);
}

10.建立接口ProducerController 测试

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;@Slf4j
@RestController
@RequestMapping("/api")
public class ProducerController {@GetMapping("/hello/{message}")public ResponseEntity<String> sayHello(@PathVariable String message) {return ResponseEntity.ok("Hello, RabbitMQ!==>"+message);}@Autowiredprivate RabbitTemplate rabbitTemplate;@GetMapping("/sendMsg/{message}")public void sendMsg(@PathVariable String message){log.info("接收接口发送的消息请求");rabbitTemplate.convertAndSend("X","XA","消息来自tt1为10s的队列"+message);rabbitTemplate.convertAndSend("X","XB","消息来自tt1为150s的队列"+message);}}

11.测试接口请求1

测试请求接口 /hello
http://实际本机ip地址:1001/demo1/api/hello/返回浏览器输入内容
在这里插入图片描述

12.测试接口请求2

测试请求接口 /hello
http://实际本机ip地址:1001/demo1/api/sendRabbitMq/发给rabbitmq消息
在这里插入图片描述

13.网络助手测试NetAssist.exe

主要目的模拟向netty端口进行发送数据，通过netty监听到的信息然后通过rabbitmq的生产者发送rabbitmq的队列中，让消费者进行消费，如果消费者绑定死信队列,那么消费者从队列中取出消息后，经过一定时间未确认即不进行消费确认或者拒绝，然后入之前绑定好的死信队列中，供死信队列绑定的死信消费者进行消费处理。

rabbitmq_578">14.观察rabbitmq界面管理简单介绍

在这里插入图片描述


http://www.ppmy.cn/news/1534482.html

相关文章

如何在Spring Boot中有条件地运行CommandLineRunner Bean

如何在Spring Boot中有条件地运行CommandLineRunner Bean

PS 使用 Spring Boot 3.1.2 进行测试 1.使用ConditionalOnProperty ConditionalOnProperty仅当特定属性存在或具有特定值时&#xff0c;注释才会创建 Bean 。 在此示例中&#xff0c;仅当或文件中的CommandLineRunner属性db.init.enabled设置为 true时&#xff0c;才会执行。…
阅读更多...
【GEE学习第三期】GEE常用函数总结

【GEE学习第三期】GEE常用函数总结

【GEE学习第三期】GEE常用函数总结 数据统计类ee.List.sequence函数 图像处理类ee.Geometry类‌defaultVisualizationVis函数 数据输入输出数值与绘图导出影像 参考 数据统计类 ee.List.sequence函数 用法如下&#xff1a; ee.List.sequence &#xff08;开始&#xff0c;结…
阅读更多...
【力扣 | SQL题 | 每日三题】力扣1264， 1113， 1098， 1082

【力扣 | SQL题 | 每日三题】力扣1264， 1113， 1098， 1082

1. 力扣1264&#xff1a;页面推荐 1.1 题目&#xff1a; 朋友关系列表&#xff1a; Friendship ------------------------ | Column Name | Type | ------------------------ | user1_id | int | | user2_id | int | ------------------------ (user…
阅读更多...
vue访问组件的数据和方法

vue访问组件的数据和方法

组件源码 <template><view class"c_container" :style"myStyle" click"clickCust"><view style"font-size: 18px;text-align: center;">{{item.name}}</view><view style"margin-top: 10px;font-siz…
阅读更多...
C++模拟实现vector容器【万字模拟✨】

C++模拟实现vector容器【万字模拟✨】

更多精彩内容..... &#x1f389;❤️播主の主页✨&#x1f618; Stark、-CSDN博客 本文所在专栏&#xff1a; 学习专栏C语言_Stark、的博客-CSDN博客 项目实战C系列_Stark、的博客-CSDN博客 数据结构与算法_Stark、的博客-CSDN博客 座右铭&#xff1a;梦想是一盏明灯&#xff…
阅读更多...
lambda表达式底层实现：反编译LambdaMetafactory + 转储dump + 运行过程 + 反汇编 + 动态指令invokedynamic

lambda表达式底层实现：反编译LambdaMetafactory + 转储dump + 运行过程 + 反汇编 + 动态指令invokedynamic

一、结论先行 lambda 底层实现机制 1.lambda 表达式的本质&#xff1a;函数式接口的匿名子类的匿名对象 2.lambda表达式是语法糖 语法糖&#xff1a;编码时是lambda简洁的表达式&#xff0c;在字节码期&#xff0c;语法糖会被转换为实际复杂的实现方式&#xff0c;含义不变&am…
阅读更多...
Arduino UNO R3自学笔记17 之 Arduino为啥要用中断？

Arduino UNO R3自学笔记17 之 Arduino为啥要用中断？

注意&#xff1a;学习和写作过程中&#xff0c;部分资料搜集于互联网&#xff0c;如有侵权请联系删除。 前言&#xff1a;学习Arduino中断的概念及其功能。 1.什么是中断&#xff1f; 单片机在执行程序时&#xff0c;发生一些其它紧急的事情&#xff0c;单片机将立即暂停当前…
阅读更多...
Python编码系列—Python访问者模式：为对象结构添加新功能的艺术

Python编码系列—Python访问者模式：为对象结构添加新功能的艺术

&#x1f31f;&#x1f31f; 欢迎来到我的技术小筑&#xff0c;一个专为技术探索者打造的交流空间。在这里&#xff0c;我们不仅分享代码的智慧&#xff0c;还探讨技术的深度与广度。无论您是资深开发者还是技术新手&#xff0c;这里都有一片属于您的天空。让我们在知识的海洋中…
阅读更多...
最新文章

Copyright @ 2022~2023