教育网站的建设,男女在床上做羞羞的事的网站,装修网站建设方案百度文库,天津定制网站建设公司一、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
