网站优化潍坊,wordpress调用侧边栏,cargo创建个人网站,郑州建设银行网站房贷网点在哪里文章目录 创建SpringBoot和RabbitMQ的整合项目首先快速创建一个maven项目引入SpringBoot整合rabbitMQ的依赖在src/main目录下创建resources目录并引入配置文件写消息发送者MessageSender写消息接收者MessageReceiver写RabbitMQConfig配置类写SpringBoot启动主类CommandLineRunn… 文章目录 创建SpringBoot和RabbitMQ的整合项目首先快速创建一个maven项目引入SpringBoot整合rabbitMQ的依赖在src/main目录下创建resources目录并引入配置文件写消息发送者MessageSender写消息接收者MessageReceiver写RabbitMQConfig配置类写SpringBoot启动主类CommandLineRunner接口的作用启动程序 查看效果注意要点15672和5672的区别消息发送的队列怎么找 我们的消息发送者和消息接收者可以只开启一个只开启消息发送者先存一些消息到消息队列只开启消息接收者从消息队列里面读取消息 接下来我们来看一下RabbitMQ中怎么控制一个线程一次消费的消息数主要是看下concurrency关键属性 创建SpringBoot和RabbitMQ的整合项目
首先快速创建一个maven项目
如下图
引入SpringBoot整合rabbitMQ的依赖
如下图 这里要SpringBoot和RabbitMQ整合的依赖对应的就是spring-boot-starter-amqp那么amqp是什么意思呢是Advanced Message Queuing Protocol高级消息队列协议是应用层协议的一个开放标准是面向消息的中间件设计。比如RabbitMQ消息中间件使用的就是这个协议。 因为我们要启动服务因此需要有tomcat服务器因此这里还要引入spring-boot-starter-web模块依赖因为SpringBoot自带的tomcat服务器在web模块里面。
project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdorg.example/groupIdartifactIdrabbitMQ/artifactIdversion1.0-SNAPSHOT/versionpackagingjar/packagingnamerabbitMQ/nameurlhttp://maven.apache.org/urlpropertiesproject.build.sourceEncodingUTF-8/project.build.sourceEncoding/propertiesparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.6.7/version/parentdependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdjunit/groupIdartifactIdjunit/artifactIdversion3.8.1/versionscopetest/scope/dependency/dependencies
/project
在src/main目录下创建resources目录并引入配置文件
如下图
写消息发送者MessageSender
如下图
/*** author xuan* create 2024/4/25*/
Component
public class MessageSender {Autowiredprivate AmqpTemplate rabbitTemplate;public void sendMessage(String exchange, String routingKey, Object message) {rabbitTemplate.convertAndSend(exchange, routingKey, message);}
}写消息接收者MessageReceiver
如下图
/*** author xuan* create 2024/4/25*/
Component
public class MessageReceiver {RabbitListener(queues test.queue)public void handleMessage(Object message) {System.out.println(Received message: message);}
}写RabbitMQConfig配置类
如下图
/*** author xuan* create 2024/4/25*/
Configuration
public class RabbitMQConfig {Beanpublic DirectExchange testExchange() {return new DirectExchange(test.exchange);}Beanpublic Queue testQueue() {return new Queue(test.queue);}Beanpublic Binding testBinding() {return BindingBuilder.bind(testQueue()).to(testExchange()).with(test.routingKey);}
}
写SpringBoot启动主类
如下图
/*** author xuan* create 2024/4/25*/
SpringBootApplication
public class Application implements CommandLineRunner {Autowiredprivate MessageSender messageSender;public static void main(String[] args) {SpringApplication.run(Application.class, args);}Overridepublic void run(String... args) throws Exception {messageSender.sendMessage(test.exchange, test.routingKey, Hello, RabbitMQ!);}
}CommandLineRunner接口的作用
从上面的SpringBoot启动类可以看到它实现了CommandLineRunner接口实现了这个接口中的抽象方法run那么CommandLineRunner接口又什么作用呢它主要是用在SpringBoot的启动主类上的主要作用是在SpringBoot应用程序启动之后执行一些任务比如
数据初始化加载初始数据到数据库或其他存储系统中。发送通知向用户或系统发送启动通知例如通过邮件或消息队列。执行检查进行应用程序启动前的检查如数据库连接检查、文件系统检查等。
比如我们上面的主类在成功启动SpringBoot的应用程序之后 就会给RabbitMQ消息队列发送一个消息。
启动程序 查看效果
启动主类之后 控制台输出如下图 这里我们主要看消息体Body的内容可以发现是Hello RabbitMQ。后面是消息的一些属性 类似如我们的http协议中的request请求头的信息包括消息的内容类型了消息的内容编码了消息的长度了等信息。
可以发现我们这里测试成功了。
注意要点
15672和5672的区别
RabbitMQ前端管理界面使用的端口号是15672比如我们启动rabbitmq之后在浏览器中输入127.0.0.1:15672可以进入rabbitmq的前端页面管理器中进行相关操作如下图 但是后端服务器的rabbitmq在电脑上启动的端口号是5672如下图 这两个端口号特别容易搞错千万要注意如果配置文件里的端口号写15672那么我们的idea里面的java程序就连接不上我们本地已经启动的rabbitmq服务器了。
消息发送的队列怎么找
通过RabbitMQ中的交换机Exchange和路由RouteingKey可以唯一的定位到一个消息队列Queue如下图
因此我们代码里面发送者发送消息的时候一定要指定使用的交换机和使用的路由但不用指定使用的队列因为有交换机和路由之后就会自动的映射到使用的队列了看下代码如下图 但是我们需要通过一个配置类配置一下 需要告诉java程序如下图 然后接收者接收消息的时候 可以直接指定test.queue队列就可以自动的去这个队列中取出消息了。不用写交换机和路由的具体信息。下面看下接收者的处理 如下图
但是我觉得RabbitMQ应该是会有线程安全问题的比如说因为目前看来同一个队列 两个不同的线程都是可以访问的那这样的话有共享资源问题 就肯定会出现线程安全问题。
我们的消息发送者和消息接收者可以只开启一个
只开启消息发送者先存一些消息到消息队列
比如我们可以先用消息发送者往RabbitMQ的消息队列test.queue里面发送三条消息此时注意关闭消息接收者以免接收者直接读取队列里面的消息我们在页面上就看不到效果了如下图
只开启消息接收者从消息队列里面读取消息
接下来我们把消息发送者关了把消息接收者打开去消费队列里面的消息如下图 看下启动应用程序之后的效果 如下图
接下来我们来看一下RabbitMQ中怎么控制一个线程一次消费的消息数主要是看下concurrency关键属性
通过concurrency属性来控制线程从队列中一次性读取的消息个数。如果我们把属性concurrency的值设置为1 那么就表示一个线程一次性会从队列里面读取一条消息如果把这个值设置成3 那么就表示一个线程一次性会从队列里面读取3条消息concurrency的书写位置如下图 据我的观察RabbitMQ内部应该有线程安全机制就是在当前线程读取消费队列消息没有完成的时候另外的线程是不能消费队列消息的。
现在我们模拟两个线程而我们具体的模拟方式其实是重新复制一个项目然后改下端口号为8081接着我们把给队列发送消息的代码注释掉 只开启接收者从消息队列里面读取消息 过程如下图
上面是我们模拟的第一个线程其实也就是copy一个相同的项目 改下端口号 启动的时候就是不同的线程了。
然后启动第二个线程发送6条消息到消息队列中如下图 那么这个时候呢线程一就会先读取消息队列的第一条消息线程二此时会在线程一后面排队当线程一处理完消息之后线程二会去读取消息队列里面的消息然后线程一读取…。因此正常情况的输出是 其中一个应用程序中 输出的是编号1,35的消息另外一个应用程序中输出的是编号2,46的消息。如下图 确实是正常的。
现在我们把concurrency线程一次性最多读取的消息个数改成4看下控制台输出结果如下图 可以看到第一个线程一次性会读取消息队列的前四条消息而第二个线程本来也会读取四条 但是队列中只剩两条消息因此第二个线程就一次性的读取两条消息了。 但是有个线程安全问题。就是你会发现第一个线程虽然读取的确实是队列里面的前四条消息但是呢消费消息的时候却不是按队列里面添加消息的顺序这可能就会出现问题了。比如假如说队列里面按照先后顺序添加了两条消息1穿装2脱装。但是线程读取消息之后 先消费处理的可能是2脱装那是不是就会出现问题了因为你玩家身上现在根本就没有装备 你要怎么脱呢因此当concurrency设置的值超过1的时候都可能会出现消息消费顺序有误的问题。
因此我们如果想要保证消费的消息严格的按照队列里面添加的消息顺序来执行那么必须要把concurrency设置成1。
