https://my.oschina.net/never/blog/140368
spring大家太熟,就不多说了
rabbitmq一个amqp的队列服务实现,具体介绍请参考本文http://lynnkong.iteye.com/blog/1699684
本文侧重介绍如何将rabbitmq整合到项目中
ps:本文只是简单一个整合介绍,属于抛砖引玉,具体实现还需大家深入研究哈..
1.首先是生产者配置
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">
<!-- 连接服务配置 -->
<rabbit:connection-factory id="connectionFactory" host="localhost" username="guest"
password="guest" port="5672" />
<rabbit:admin connection-factory="connectionFactory"/>
<!-- queue 队列声明-->
<rabbit:queue id="queue_one" durable="true" auto-delete="false" exclusive="false" name="queue_one"/>
<!-- exchange queue binging key 绑定 -->
<rabbit:direct-exchange name="my-mq-exchange" durable="true" auto-delete="false" id="my-mq-exchange">
<rabbit:bindings>
<rabbit:binding queue="queue_one" key="queue_one_key"/>
</rabbit:bindings>
</rabbit:direct-exchange>
<-- spring amqp默认的是jackson 的一个插件,目的将生产者生产的数据转换为json存入消息队列,由于fastjson的速度快于jackson,这里替换为fastjson的一个实现 -->
<bean id="jsonMessageConverter" class="mq.convert.FastJsonMessageConverter"></bean>
<-- spring template声明-->
<rabbit:template exchange="my-mq-exchange" id="amqpTemplate" connection-factory="connectionFactory" message-converter="jsonMessageConverter"/>
</beans>
2.fastjson messageconver插件实现
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.AbstractMessageConverter;
import org.springframework.amqp.support.converter.MessageConversionException;
import fe.json.FastJson;
public class FastJsonMessageConverter extends AbstractMessageConverter {
private static Log log = LogFactory.getLog(FastJsonMessageConverter.class);
public static final String DEFAULT_CHARSET = "UTF-8";
private volatile String defaultCharset = DEFAULT_CHARSET;
public FastJsonMessageConverter() {
super();
//init();
}
public void setDefaultCharset(String defaultCharset) {
this.defaultCharset = (defaultCharset != null) ? defaultCharset
: DEFAULT_CHARSET;
}
public Object fromMessage(Message message)
throws MessageConversionException {
return null;
}
public <T> T fromMessage(Message message,T t) {
String json = "";
try {
json = new String(message.getBody(),defaultCharset);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
return (T) FastJson.fromJson(json, t.getClass());
}
protected Message createMessage(Object objectToConvert,
MessageProperties messageProperties)
throws MessageConversionException {
byte[] bytes = null;
try {
String jsonString = FastJson.toJson(objectToConvert);
bytes = jsonString.getBytes(this.defaultCharset);
} catch (UnsupportedEncodingException e) {
throw new MessageConversionException(
"Failed to convert Message content", e);
}
messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
messageProperties.setContentEncoding(this.defaultCharset);
if (bytes != null) {
messageProperties.setContentLength(bytes.length);
}
return new Message(bytes, messageProperties);
}
}
3.生产者端调用
import java.util.List;
import org.springframework.amqp.core.AmqpTemplate;
public class MyMqGatway {
@Autowired
private AmqpTemplate amqpTemplate;
public void sendDataToCrQueue(Object obj) {
amqpTemplate.convertAndSend("queue_one_key", obj);
}
}
4.消费者端配置(与生产者端大同小异)
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">
<!-- 连接服务配置 -->
<rabbit:connection-factory id="connectionFactory" host="localhost" username="guest"
password="guest" port="5672" />
<rabbit:admin connection-factory="connectionFactory"/>
<!-- queue 队列声明-->
<rabbit:queue id="queue_one" durable="true" auto-delete="false" exclusive="false" name="queue_one"/>
<!-- exchange queue binging key 绑定 -->
<rabbit:direct-exchange name="my-mq-exchange" durable="true" auto-delete="false" id="my-mq-exchange">
<rabbit:bindings>
<rabbit:binding queue="queue_one" key="queue_one_key"/>
</rabbit:bindings>
</rabbit:direct-exchange>
<!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象-->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto" task-executor="taskExecutor">
<rabbit:listener queues="queue_one" ref="queueOneLitener"/>
</rabbit:listener-container>
</beans>
5.消费者端调用
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
public class QueueOneLitener implements MessageListener{
@Override
public void onMessage(Message message) {
System.out.println(" data :" + message.getBody());
}
}
6.由于消费端当队列有数据到达时,对应监听的对象就会被通知到,无法做到批量获取,批量入库,因此可以在消费端缓存一个临时队列,将mq取出来的数据存入本地队列,后台线程定时批量处理即可
分享到:
相关推荐
Spring boot整合消息队列RabbitMQ
本项目基于Spring的AMQP模块,整合流行的开源消息队列中间件rabbitMQ,实现一个向rabbitMQ添加和读取消息的功能。并比较了两种模式:生产者-消费者模式和发布-订阅模式的区别。AMQP作为比JMS更加高级的消息协议,支持...
本项目基于Spring的AMQP模块,整合流行的开源消息队列中间件rabbitMQ,实现一个向rabbitMQ添加和读取消息的功能。并比较了两种模式:生产者-消费者模式和发布-订阅模式的区别。AMQP作为比JMS更加高级的消息协议,支持...
70、秒杀系统高并发之消息队列RabbitMQ和代码编写 71、秒杀系统高并发之RabbitMQ和spring整合
springcloud bus rabbitmq 分布式队列 http://knight-black-bob.iteye.com/blog/2356839
https://www.jianshu.com/p/3841059f7ca3 整合spring和rabbitmq
百度spring整合activemq 发现几乎都只是在xml文件配置固定的消息队列而且太麻烦。并没有根据需求进行动态生成主题和队列。本文档参考了纯粹的activemq java代码和百度上的demo,很简洁的实现了动态消息队列的生成和...
本篇文章主要介绍了消息队列 RabbitMQ 与 Spring 整合使用的实例代码,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
本篇文章主要介绍了rabbitmq结合spring实现消息队列优先级的方法,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
本项目基于Spring的AMQP模块,整合流行的开源消息队列中间件rabbitMQ,实现一个向rabbitMQ添加和读取消息的功能。并比较了两种模式:生产者-消费者模式和发布-订阅模式的区别。AMQP作为比JMS更加高级的消息协议,支持...
Spring项目集成了消息队列-RabbitMQ,考虑到数据安全和稳定性,选择了RabbitMQ.当前开发环境在Win10下测试 内置功能 基于普通消息订阅-发布模式的消息队列实现。 基于延迟消息队列的实现。 基于定时消息发送的功能...
spring集成rabbitmq 通俗易懂的demo,保证可以使用,具体的配置,可以参考我的文章
springboot集成消息队列实现发送与接收demo; 包含activemq rocketmq kafka等
spring boot 配置, Rabbitmq集成, 利用死信 实现延时消息队列实现
mysql需开启binlog 查看是否开启binlog ... 新增队列:test.queue, 绑定canal.queue, RoutingKey:canal.routing.key canal下载及配置 https://github.com/alibaba/canal/releases/tag/canal-1.1.5 配置文件见附件
积分管理系统java源码 ...3)与SpringAMQP完美整合,API丰富 4)集群模式丰富,表达式配置,HA模式,镜像队列模型 5)保证数据不丢失的前提下做到高可靠性、可用性 RabbitMQ安装与使用 Window 安装Erlang 去官
该示例通过 rabbitmq_delayed_message_exchange 插件实现自定义延时时间的延时队列。 示例是纯净的,只引入了需要的架包 启动示例时,请确保MQ已经安装了延时插件(附件里带有插件及安装说明)以及示例的MQ相关的配置...
RabbitMQ 是一个开源的消息代理中间件,广泛用于构建分布式应用程序中的消息系统。在 Spring Boot 项目中,通过集成 RabbitMQ,可以实现异步消息传递、消息队列等功能,提高系统的可靠性和扩展性。
springboot整合rabbitmq使用死信队列
网络视频资源,如有侵权请留言/举报,资源过大上传乃是下载链接!!! 『课程介绍』: Rabbitmq企业级的消息服务系统,可以在应用间收发消息,本课程主要学习...10、整合spring方式-1 11、整合sring方式-2 12、消息持久化