靖州网站建设,兴趣电商平台有哪些,有些网站为什么可以做资讯,栏目手机版 企业网站文章目录一、简单对象1. 依赖2. 生产者3. 消费者4. 配置文件5. spring版本二、复杂对象2.1. 生产者2.2. 消费者一、简单对象
1. 依赖 !--spring整合rabbitmq--dependencygroupIdorg.springframework.amqp/groupIdartifactIdspring-ra…
文章目录一、简单对象1. 依赖2. 生产者3. 消费者4. 配置文件5. spring版本二、复杂对象2.1. 生产者2.2. 消费者一、简单对象
1. 依赖 !--spring整合rabbitmq--dependencygroupIdorg.springframework.amqp/groupIdartifactIdspring-rabbit/artifactIdversion2.0.1.RELEASE/version/dependency注maven方式这一个依赖即可如果是非maven项目需要引入5个jar如下 推荐使用mavne方式简单非Maven项目先用maven把以来下载本地仓库复制到非maven的项目中即可。
2. 生产者
?xml version1.0 encodingUTF-8?
beans xmlnshttp://www.springframework.org/schema/beansxmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexmlns:rabbithttp://www.springframework.org/schema/rabbitxsi:schemaLocationhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit-2.0.xsdhttp://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans-3.1.xsd!--生产者者配置如下--!-- 定义RabbitMQ的连接工厂 --rabbit:connection-factory idconnectionFactoryhost${rabbitmq.host}port${rabbitmq.port}username${rabbitmq.username}password${rabbitmq.password}virtual-host${rabbitmq.vhost}connection-timeout${rabbitmq.conTimeout}publisher-confirms${rabbitmq.publisher-confirms}publisher-returns${rabbitmq.publisher-returns}/!-- 管理消息队列 --rabbit:admin connection-factoryconnectionFactory/!--此处为配置文件方式 管控台配置模式需要注释 默认模式管控台 Start--!-- 定义一个队列或者多个队列 自动声明--rabbit:queue nameQueue-1 auto-declaretrue durabletrue/rabbit:topic-exchange nameexchange-1rabbit:bindings!-- 可绑定多个队列发送的时候指定key进行发送 --rabbit:binding queueQueue-1 patternws.tjqb//rabbit:bindings/rabbit:topic-exchange!--此处为配置文件方式 管控台配置模式需要注释 默认模式管控台 End--!-- 定义交换机 自动声明--rabbit:topic-exchange nameexchange-1auto-declaretrue durabletrue/!-- 5. 配置消息对象json转换类 --bean idjsonMessageConverter classorg.springframework.amqp.support.converter.Jackson2JsonMessageConverter /!-- 定义MQ消息模板1. id 定义消息模板ID2.connection-factory 把定义的连接工厂放到消息模板中3.confirm-callback confirm确认机制4.return-callback return确认机制5.mandatory #有2种状态设置为 true 后 消费者在消息没有被路由到合适队列情况下会被return监听而不会自动删除设置为 false 后 消费者在消息没有被路由到合适队列情况下会自动删除--rabbit:template idrabbitTemplateconnection-factoryconnectionFactoryexchangeexchange-1confirm-callbackconfirmCallBackListenerreturn-callbackreturnCallBackListenermandatorytruemessage-converterjsonMessageConverter/
/beanspackage com.gblfy.order.controller;import com.gblfy.order.pojo.FisCallingTrace;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;RestController
public class Send {public static final String EXCHANGE exchange-1;AutowiredRabbitTemplate rabbitTemplate;RequestMapping(/test)public String test() {String uuidStr UUID.randomUUID().toString();CorrelationData correlationId new CorrelationData(uuidStr);// 发送消息MapString, String map new HashMap();map.put(email, 550731230qq.com);rabbitTemplate.convertAndSend(EXCHANGE, ws.tjqb, map, correlationId);return success;}3. 消费者
?xml version1.0 encodingUTF-8?
beans xmlnshttp://www.springframework.org/schema/beansxmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexmlns:rabbithttp://www.springframework.org/schema/rabbitxsi:schemaLocationhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit-2.0.xsdhttp://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans-3.1.xsd!--消费者配置如下--!-- 定义RabbitMQ的连接工厂 --rabbit:connection-factory idconnectionFactoryhost${rabbitmq.host} port${rabbitmq.port} username${rabbitmq.username}password${rabbitmq.password} virtual-host${rabbitmq.vhost}connection-timeout${rabbitmq.conTimeout}publisher-confirms${rabbitmq.publisher-confirms}publisher-returns${rabbitmq.publisher-returns}/!-- 管理消息队列 --rabbit:admin connection-factoryconnectionFactory/!-- 声明多个消费者对象 --bean idemailMessageListener classcom.gblfy.order.mqhandler.EmailMessageListener/!-- 监听队列1. connectionFactory 连接工厂2. manual 手动签收3. ref 消费者监听--rabbit:listener-container connection-factoryconnectionFactoryacknowledgemanualconcurrency${rabbitmq.concurrency}max-concurrency${rabbitmq.max-concurrency}rabbit:listener refemailMessageListener methodonMessage queue-namesQueue-1//rabbit:listener-container
/beans
package com.gblfy.order.mqhandler;import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;import java.io.IOException;
Component
public class EmailMessageListener implements MessageListener {private static final ObjectMapper MAPPER new ObjectMapper();Overridepublic void onMessage(Message message) {try {JsonNode jsonNode MAPPER.readTree(message.getBody());String email jsonNode.get(email).asText();System.out.println(获取队列中消息 email);} catch (IOException e) {e.printStackTrace();}}
}4. 配置文件
#RabbitMQ 连接信息
#IP地址
rabbitmq.host192.168.0.114
#端口
rabbitmq.port5672
#用户名
rabbitmq.usernamefis
#密码
rabbitmq.passwordncl1234
#虚拟主机
rabbitmq.vhost/app/fisMQ
#连接超时时间
rabbitmq.conTimeout15000
#发送确认 对应RabbitTemplate.ConfirmCallback接口
#消息发送成功 有2个重要参数
# ack 状态为true correlationId 全局唯一ID用于标识每一支队列
rabbitmq.publisher-confirmstrue
#发送失败回退对应RabbitTemplate.ReturnCallback接口
rabbitmq.publisher-returnstrue
#默认消费者数量
rabbitmq.concurrency10
#最大消费者数量
rabbitmq.max-concurrency20
5. spring版本
目前适配的spring版本4.2.3.RELEASE
二、复杂对象
声明配置文件不变
2.1. 生产者
package com.gblfy.order.controller;import com.gblfy.order.pojo.FisCallingTrace;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;RestController
public class Send {public static final String EXCHANGE exchange-1;AutowiredRabbitTemplate rabbitTemplate;RequestMapping(/test)public String test() {FisCallingTrace f getFisCallingTrace();String uuidStr UUID.randomUUID().toString();CorrelationData correlationId new CorrelationData(uuidStr);// 发送消息MapString, Object map new HashMap();map.put(mReqXml, 请求报文);map.put(mResXml, 响应报文);map.put(mUUID, uuidStr);map.put(serviceName, NYHC);map.put(fisCallingTrace, f);rabbitTemplate.convertAndSend(EXCHANGE, ws.tjqb, map, correlationId);return success;}// RequestMapping(/test)// public String test() {// String uuidStr UUID.randomUUID().toString();// CorrelationData correlationId new CorrelationData(uuidStr);// // 发送消息// MapString, String map new HashMap();// map.put(email, 550731230qq.com);// rabbitTemplate.convertAndSend(EXCHANGE, ws.tjqb, map, correlationId);// return success;// }private FisCallingTrace getFisCallingTrace() {FisCallingTrace f new FisCallingTrace();f.setServicename(tjqb);f.setServicetype(1);f.setInterfacetype(2);f.setResstatus(1);f.setResremark(纽约数据回传接口);f.setReqdate(new Date());f.setReqtime(10:00:00);f.setResdate(new Date());f.setRestime(10:00:00);f.setReqxml(请求报文);f.setResxml(响应报文);return f;}}
2.2. 消费者
package com.gblfy.order.mqhandler;import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.gblfy.order.pojo.FisCallingTrace;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;import java.io.IOException;Slf4j
Component
public class EmailMessageListener implements ChannelAwareMessageListener {private static final ObjectMapper MAPPER new ObjectMapper();Overridepublic void onMessage(Message message, Channel channel) throws Exception {try {JsonNode jsonNode MAPPER.readTree(message.getBody());String mReqXml jsonNode.get(mReqXml).asText();String mResXml jsonNode.get(mResXml).asText();String mUUID jsonNode.get(mUUID).asText();String serviceName jsonNode.get(serviceName).asText();System.out.println(获取队列中消息 mReqXml);System.out.println(获取队列中消息 mResXml);System.out.println(获取队列中消息 mUUID);System.out.println(获取队列中消息 serviceName);JsonNode jsonNode1 jsonNode.get(fisCallingTrace);String jsonStr MAPPER.writeValueAsString(jsonNode1);FisCallingTrace f MAPPER.readValue(jsonStr , FisCallingTrace.class);System.out.println(获取队列中消息 f.getReqxml());System.out.println(获取队列中消息 f.getResxml());// 消息的标识false只确认当前一个消息收到true确认所有consumer获得的消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (IOException e) {e.printStackTrace();}log.info(解析操作);log.info(落库操作);}
}
