网站品牌建设公司,网站开发流程框架,网站建设盈利,帝国cms网站地图生成器一、pom依赖二、消费端2.1、application.properties 配置文件2.2、消费端核心组件 三、生产端3.1、application.properties 配置文件2.2、生产者 MQ消息发送组件四、测试1、生产端控制台2、消费端控制台 一、pom依赖 dependencygroupIdorg.springframework.boo… 一、pom依赖二、消费端2.1、application.properties 配置文件2.2、消费端核心组件 三、生产端3.1、application.properties 配置文件2.2、生产者 MQ消息发送组件四、测试1、生产端控制台2、消费端控制台 一、pom依赖 dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependency!--spring整合MQ--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactId/dependency二、消费端
2.1、application.properties 配置文件
server.port8002
#上下文路径
server.servlet.context-path/
spring.application.namerabbit_consumer# MQ配置
spring.rabbitmq.addresses192.168.220.3:5672
spring.rabbitmq.usernameroot
spring.rabbitmq.passwordroot
# 虚拟主机
spring.rabbitmq.virtual-host/
# 连接超时 15秒
spring.rabbitmq.connection-timeout15000
# 设置消费端消费成功消息后手动签收消息默认auto自动签收
spring.rabbitmq.listener.simple.acknowledge-modemanual
spring.rabbitmq.listener.simple.concurrency6
# 最大消费线程数并发数
spring.rabbitmq.listener.simple.max-concurrency11
# prefetch为限制一次传送给消费者的消息数
spring.rabbitmq.listener.simple.prefetch1# 自定义属性配置 MQ
spring.rabbitmq.listener.test.exchangetest_topic_exchange
spring.rabbitmq.listener.test.exchange.typetopic
spring.rabbitmq.listener.test.queuetest_topic1
spring.rabbitmq.listener.test.keytest_topic1.*2.2、消费端核心组件
package com.xiao.component;import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import java.io.IOException;Component
public class RabbitMQReceived {RabbitListener(bindings QueueBinding(exchange Exchange(name ${spring.rabbitmq.listener.test.exchange},type ${spring.rabbitmq.listener.test.exchange.type},durable true,ignoreDeclarationExceptions true),value Queue(value ${spring.rabbitmq.listener.test.queue},durable true),key ${spring.rabbitmq.listener.test.key}/*,admins root*/))/*** 监听消息* param message 消息* param channel 通道*/RabbitHandlerpublic void onMessage(Message message, Channel channel) throws IOException {System.err.println();System.err.println(消费端 RabbitMQReceived 消费消息 message.getPayload());Long deliveryTag (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);//由于消费端配置手动消费消息后签收机制 spring.rabbitmq.listener.simple.acknowledge-modemanual
// channel.basicAck(deliveryTag,false);System.err.println(消费端 RabbitMQReceived ackyes deliveryTag: deliveryTag);}
}三、生产端
3.1、application.properties 配置文件
server.port8001
#上下文路径
server.servlet.context-path/
spring.application.namerabbit_produce# MQ配置
spring.rabbitmq.addresses192.168.220.3:5672
spring.rabbitmq.port5672
spring.rabbitmq.usernameroot
spring.rabbitmq.passwordroot
# 虚拟主机
spring.rabbitmq.virtual-host/
# 连接超时 15秒
spring.rabbitmq.connection-timeout15000
# 开启produce发送给broker的消息确认模式可靠性投递
spring.rabbitmq.publisher-confirmstrue
#spring.rabbitmq.publisher-confirm-typetrue #有点问题
# 针对于broker未接收的消息return机制需要结合mandatory一起使用
#spring.rabbitmq.template.mandatorytrue
#spring.rabbitmq.publisher-returnstrue2.2、生产者 MQ消息发送组件
package com.xiao.component;import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Correlation;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.UUID;Component
public class RabbitMQSender {Autowiredprivate RabbitTemplate rabbitTemplate;//生产者发送消息到broker确认回调接口private final RabbitTemplate.ConfirmCallback confirmCallback new RabbitTemplate.ConfirmCallback() {/*** param correlationData 消息的唯一标识* param ack broke broker是否签收成功* param cause 失败异常信息*/Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {String formatStr String.format(生产端 confirmCallback 相关数据%s broker签收情况 ack%s异常信息%s ,correlationData.toString(),ack,cause);System.err.println(formatStr);/*System.out.println(生产端 confirmCallback 相关数据 correlationData);System.out.println(生产端 confirmCallback broker签收情况 ack);System.out.println(生产端 confirmCallback 异常信息 cause);*/}};/*** 发送消息* param message 消息* param properties 消息对应的属性如时间*/public void send(Object message, MapString,Object properties) {MessageHeaders messageHeaders new MessageHeaders(properties);Message? msg MessageBuilder.createMessage(message, messageHeaders);rabbitTemplate.setConfirmCallback(confirmCallback);//消息发送完后置处理器MessagePostProcessor messagePostProcessor new MessagePostProcessor() {Overridepublic org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {System.err.println(生产端 RabbitMQSender send后置处理 message);return message;}Overridepublic org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message, Correlation correlation) {System.err.println(生产端 RabbitMQSender send后置处理 message 消息标识 correlation);return message;}};//消息唯一属性CorrelationData correlationData new CorrelationData(UUID.randomUUID().toString());rabbitTemplate.convertAndSend(test_topic_exchange,//exchange,test_topic1.xiao,// routingKey,msg, //message,messagePostProcessor,correlationData);}
}四、测试
package com.xiao;import com.xiao.component.RabbitMQSender;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import java.util.HashMap;
import java.util.Map;RunWith(SpringJUnit4ClassRunner.class)
SpringBootTest
public class SendMessageTest {Autowiredprivate RabbitMQSender rabbitMQSender;Testpublic void send() throws InterruptedException {MapString,Object properties new HashMap(2);properties.put(userName,xiao);rabbitMQSender.send(hello world!,properties);Thread.sleep(5000);//10秒}
}1、生产端控制台
生产端 RabbitMQSender send后置处理(Body:[B3a6045c6(byte[535]) MessageProperties [headers{}, contentTypeapplication/x-java-serialized-object, contentLength535, deliveryModePERSISTENT, priority0, deliveryTag0]) 消息标识CorrelationData [id8c78e89d-80f3-4f3d-ba8b-13e863c6295c]
2023-07-21 20:05:37.611 INFO 4536 --- [ main] o.s.a.r.c.CachingConnectionFactory : Attempting to connect to: [192.168.220.3:5672]
2023-07-21 20:05:37.653 INFO 4536 --- [ main] o.s.a.r.c.CachingConnectionFactory : Created new connection: rabbitConnectionFactory#6f38a289:0/SimpleConnection6215366a [delegateamqp://root192.168.220.3:5672/, localPort 4712]
生产端 confirmCallback 相关数据CorrelationData [id8c78e89d-80f3-4f3d-ba8b-13e863c6295c]broker签收情况 acktrue异常信息null2、消费端控制台 消费端 RabbitMQReceived 消费消息hello world!
消费端 RabbitMQReceived ackyes deliveryTag:1 文章转载自： http://www.morning.sryhp.cn.gov.cn.sryhp.cn http://www.morning.wqbbc.cn.gov.cn.wqbbc.cn http://www.morning.ntqjh.cn.gov.cn.ntqjh.cn http://www.morning.ntyks.cn.gov.cn.ntyks.cn http://www.morning.qzmnr.cn.gov.cn.qzmnr.cn http://www.morning.rhjsx.cn.gov.cn.rhjsx.cn http://www.morning.zzhqs.cn.gov.cn.zzhqs.cn http://www.morning.dtzsm.cn.gov.cn.dtzsm.cn http://www.morning.kdldx.cn.gov.cn.kdldx.cn http://www.morning.kghhl.cn.gov.cn.kghhl.cn http://www.morning.rywn.cn.gov.cn.rywn.cn http://www.morning.mpflb.cn.gov.cn.mpflb.cn http://www.morning.htbgz.cn.gov.cn.htbgz.cn http://www.morning.jcnmy.cn.gov.cn.jcnmy.cn http://www.morning.qxlxs.cn.gov.cn.qxlxs.cn http://www.morning.pqqxc.cn.gov.cn.pqqxc.cn http://www.morning.ybgyz.cn.gov.cn.ybgyz.cn http://www.morning.jphxt.cn.gov.cn.jphxt.cn http://www.morning.jklns.cn.gov.cn.jklns.cn http://www.morning.pjzcp.cn.gov.cn.pjzcp.cn http://www.morning.zqcdl.cn.gov.cn.zqcdl.cn http://www.morning.hsgxj.cn.gov.cn.hsgxj.cn http://www.morning.lkhgq.cn.gov.cn.lkhgq.cn http://www.morning.wrlxt.cn.gov.cn.wrlxt.cn http://www.morning.zdfrg.cn.gov.cn.zdfrg.cn http://www.morning.lbggk.cn.gov.cn.lbggk.cn http://www.morning.rnht.cn.gov.cn.rnht.cn http://www.morning.srgsb.cn.gov.cn.srgsb.cn http://www.morning.nkyc.cn.gov.cn.nkyc.cn http://www.morning.bqrd.cn.gov.cn.bqrd.cn http://www.morning.ntqjh.cn.gov.cn.ntqjh.cn http://www.morning.dkbgg.cn.gov.cn.dkbgg.cn http://www.morning.zxybw.cn.gov.cn.zxybw.cn http://www.morning.xdjwh.cn.gov.cn.xdjwh.cn http://www.morning.kpgbz.cn.gov.cn.kpgbz.cn http://www.morning.gslz.com.cn.gov.cn.gslz.com.cn http://www.morning.ypdhl.cn.gov.cn.ypdhl.cn http://www.morning.jjsxh.cn.gov.cn.jjsxh.cn http://www.morning.fhxrb.cn.gov.cn.fhxrb.cn http://www.morning.hmgqy.cn.gov.cn.hmgqy.cn http://www.morning.jbfzx.cn.gov.cn.jbfzx.cn http://www.morning.fcxt.cn.gov.cn.fcxt.cn http://www.morning.dpdns.cn.gov.cn.dpdns.cn http://www.morning.tfpbm.cn.gov.cn.tfpbm.cn http://www.morning.pltbd.cn.gov.cn.pltbd.cn http://www.morning.xlxmy.cn.gov.cn.xlxmy.cn http://www.morning.xhpnp.cn.gov.cn.xhpnp.cn http://www.morning.mnqz.cn.gov.cn.mnqz.cn http://www.morning.pqhgn.cn.gov.cn.pqhgn.cn http://www.morning.cfcpb.cn.gov.cn.cfcpb.cn http://www.morning.qjfkz.cn.gov.cn.qjfkz.cn http://www.morning.wwjft.cn.gov.cn.wwjft.cn http://www.morning.mgfnt.cn.gov.cn.mgfnt.cn http://www.morning.kzhxy.cn.gov.cn.kzhxy.cn http://www.morning.wpmqq.cn.gov.cn.wpmqq.cn http://www.morning.rzbcz.cn.gov.cn.rzbcz.cn http://www.morning.ltcnd.cn.gov.cn.ltcnd.cn http://www.morning.qstkk.cn.gov.cn.qstkk.cn http://www.morning.dytqf.cn.gov.cn.dytqf.cn http://www.morning.lsbjj.cn.gov.cn.lsbjj.cn http://www.morning.yggdq.cn.gov.cn.yggdq.cn http://www.morning.pbsqr.cn.gov.cn.pbsqr.cn http://www.morning.fflnw.cn.gov.cn.fflnw.cn http://www.morning.xhfky.cn.gov.cn.xhfky.cn http://www.morning.frsxt.cn.gov.cn.frsxt.cn http://www.morning.khzml.cn.gov.cn.khzml.cn http://www.morning.qfmns.cn.gov.cn.qfmns.cn http://www.morning.ltypx.cn.gov.cn.ltypx.cn http://www.morning.rdxp.cn.gov.cn.rdxp.cn http://www.morning.bcjbm.cn.gov.cn.bcjbm.cn http://www.morning.tkrdg.cn.gov.cn.tkrdg.cn http://www.morning.wphfl.cn.gov.cn.wphfl.cn http://www.morning.zwppm.cn.gov.cn.zwppm.cn http://www.morning.snbq.cn.gov.cn.snbq.cn http://www.morning.zrkp.cn.gov.cn.zrkp.cn http://www.morning.rwcw.cn.gov.cn.rwcw.cn http://www.morning.czzpm.cn.gov.cn.czzpm.cn http://www.morning.yhplt.cn.gov.cn.yhplt.cn http://www.morning.zbjfq.cn.gov.cn.zbjfq.cn http://www.morning.cprbp.cn.gov.cn.cprbp.cn
