这篇文章主要介绍了简单了解如何在spring中使用RabbitMQ,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
常见的消息中间件产品:
(1)ActiveMQ
ActiveMQ 是Apache出品,最流行的,能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。
(2)RabbitMQ
AMQP协议的领导实现,支持多种场景。淘宝的MySQL集群内部有使用它进行通讯,OpenStack开源云平台的通信组件,最先在金融行业得到运用。我们在本次课程中介绍 RabbitMQ的使用。
(3)ZeroMQ
史上最快的消息队列系统
(4)Kafka
Apache下的一个子项目 。特点:高吞吐,在一台普通的服务器上既可以达到10W/s的吞吐速率;完全的分布式系统。适合处理海量数据。
(5)RocketMQ 阿里巴巴
消息中间件利用高效可靠的消息传递机制进行平台无关的数据交流,并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型,它可以在分布式环境下扩展进程间的通信。对于消息中间件,常见的角色大致也就有Producer(生产者)、Consumer(消费者)。
消息队列中间件是分布式系统中重要的组件,主要解决应用解耦,异步消息,流量削锋等问题,实现高性能,高可用,可伸缩和最终一致性架构。
Spring-amqp是对AMQP协议的抽象实现,而spring-rabbit 是对协议的具体实现,也是目前的唯一实现。底层使用的就是RabbitMQ。
已经配置好了ssm的开发环境
1.导入依赖
<dependencies> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.5.3</version> </dependency> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>2.1.3.RELEASE</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.9.5</version> </dependency> </dependencies>
2.编写生产者
2.1配置文件
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit" xsi:schemaLocation="http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.4.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.1.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> <context:component-scan base-package="cn.test.rabbitmq.spring"/> <!-- 配置连接工厂 --> <rabbit:connection-factory id="connectionFactory" virtual-host="/saas" host="127.0.0.1" port="5672" username="saas" password="saas" /> <!-- 定义mq管理 --> <rabbit:admin connection-factory="connectionFactory" /> <!-- 声明队列 --> <rabbit:queue name="spring.test.queue" auto-declare="true" durable="true" /> <!-- 定义交换机绑定队列(路由模式) --> <rabbit:direct-exchange name="spring.test.exchange"> <rabbit:bindings> <rabbit:binding queue="spring.test.queue" key="user.insert" /> </rabbit:bindings> </rabbit:direct-exchange> <!-- 定义交换机绑定队列(路由模式)使用匹配符 <rabbit:topic-exchange id="springTestExchange" name="spring.test.exchange"> <rabbit:bindings> <rabbit:binding queue="spring.test.queue" pattern="#.#" /> </rabbit:bindings> </rabbit:topic-exchange> --> <!-- 消息对象json转换类 --> <bean id="jsonMessageConverter" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" /> <!-- 定义模版 --> <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory" exchange="spring.test.exchange" message-converter="jsonMessageConverter"/> </beans>
2.2 发送方代码
这里是往RabbitMQ队列中放入任务,让消费者去取
package cn.test.rabbitmq.spring; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @Component public class MqSender { @Autowired private AmqpTemplate amqpTemplate; public void sendMessage(){ //根据key发送到对应的队列 amqpTemplate.convertAndSend("user.insert","spring整合RabbitMQ消息"); System.out.println("发送成功........"); } }
2.3 测试代码
package cn.test.rabbitmq.spring; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.test.context.junit4.SpringRunner; import java.util.HashMap; @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration(locations = "classpath:applicationContext-mq-send.xml") public class MqSendDemo { @Autowired private MqSender mqSender; @Test public void test(){ //根据key发送到对应的队列 mqSender.sendMessage(); } }
3.编写消费者
3.1 配置文件
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit" xsi:schemaLocation="http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> <!-- 配置连接工厂 --> <rabbit:connection-factory id="connectionFactory" virtual-host="/saas" host="127.0.0.1" port="5672" username="saas" password="saas" /> <!-- 定义mq管理 --> <rabbit:admin connection-factory="connectionFactory" /> <!-- 声明队列 --> <rabbit:queue name="spring.test.queue" auto-declare="true" durable="true" /> <!-- 定义消费者 --> <bean id="testMqListener" class="cn.test.rabbitmq.spring.MqListener" /> <!-- 定义消费者监听队列 --> <rabbit:listener-container connection-factory="connectionFactory"> <rabbit:listener ref="testMqListener" queues="spring.test.queue" /> </rabbit:listener-container> </beans>
3.2 监听代码
package cn.test.rabbitmq.spring; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageListener; import org.springframework.stereotype.Component; import java.io.UnsupportedEncodingException; public class MqListener implements MessageListener { public void onMessage(Message message) { try { System.out.println(message.getBody()); String ms = new String(message.getBody(), "UTF-8"); System.out.println(ms); } catch (Exception e) { e.printStackTrace(); } } }
3.3 测试代码
package cn.itcast.rabbitmq.spring; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration(locations = "classpath:applicationContext-mq-receive.xml") public class MqReceiveDemo { @Test public void test(){ //等待队列中放入任务,如果有任务,立即消费任务 while (true){ } } }
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持易盾网络。