简单了解如何在spring中使用RabbitMQ

简单了解如何在spring中使用RabbitMQ

这篇文章主要介绍了简单了解如何在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.导入依赖

com.rabbitmq

amqp-client

5.5.3

&DNxmXlt;groupId>org.springframework.amqp

spring-rabbit

2.1.3.RELEASE

com.fasterxml.jackson.core

jackson-databind

2.9.5

2.编写生产者

2.1配置文件

xmlns:context="http://springframework.org/schema/context"

xmlns:xsi="http://w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://springframework.org/schema/rabbit"

xsi:schemaLocation="http://springframework.org/schema/rabbit

http://springframework.org/schema/rabbit/spring-rabbit-1.4.xsd

http://springframework.org/schema/beans

http://springframework.org/schema/beans/spring-beans-4.1.xsd

http://springframework.org/schema/context

http://springframework.org/schema/context/spring-context.xsd">

host="127.0.0.1" port="5672" username="saas" password="saas" />

class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" />

exchange="spring.test.exchange"

message-converter="jsonMessageConverter"/>

xmlns:context="http://springframework.org/schema/context"

xmlns:xsi="http://w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://springframework.org/schema/rabbit"

xsi:schemaLocation="http://springframework.org/schema/rabbit

http://springframework.org/schema/rabbit/spring-rabbit-1.4.xsd

http://springframework.org/schema/beans

http://springframework.org/schema/beans/spring-beans-4.1.xsd

http://springframework.org/schema/context

http://springframework.org/schema/context/spring-context.xsd">

host="127.0.0.1" port="5672" username="saas" password="saas" />

class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" />

exchange="spring.test.exchange"

message-converter="jsonMessageConverter"/>

host="127.0.0.1" port="5672" username="saas" password="saas" />

class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" />

exchange="spring.test.exchange"

message-converter="jsonMessageConverter"/>

class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" />

exchange="spring.test.exchange"

message-converter="jsonMessageConverter"/>

exchange="spring.test.exchange"

message-converter="jsonMessageConverter"/>

2.2 发送方代码

这里是往RabbitMQ队列中放入任务，让消费者去取

package cn.test.rabbitmq.spring;

import org.springframework.amqp.core.AmqpTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotyphttp://e.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 配置文件

xmlns:xsi="http://w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://springframework.org/schema/rabbit"

xsi:schemaLocation="http://springframework.org/schema/rabbit

http://springframework.org/schema/rabbit/spring-rabbit.xsd

http://springframework.org/schema/beans

http://springframework.org/schema/beans/spring-beans.xsd">

host="127.0.0.1" port="5672" username="saas" password="saas" />

connection-factory="connectionFactory">

xmlns:xsi="http://w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://springframework.org/schema/rabbit"

xsi:schemaLocation="http://springframework.org/schema/rabbit

http://springframework.org/schema/rabbit/spring-rabbit.xsd

http://springframework.org/schema/beans

http://springframework.org/schema/beans/spring-beans.xsd">

host="127.0.0.1" port="5672" username="saas" password="saas" />

connection-factory="connectionFactory">

host="127.0.0.1" port="5672" username="saas" password="saas" />

connection-factory="connectionFactory">

connection-factory="connectionFactory">

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){

}

}

}

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。