Spring整合Rabbitmq

没有找到一篇完整的文章介绍Spring如何整合Rabbitmq应用,琢磨一天搞出的一个入门的demo与伙伴们分享.

第一步linux环境下安装rabbitMQ,小弟用的是ubantu,不想浪费太多时间这种安装上

sudo apt-get install rabbitmq-server

默认启动端口5672 测试账户guest 密码guest

官方常用命令,想更改用户的可以参考,这里不作详细说明http://www.rabbitmq.com/man/rabbitmqctl.1.man.html

也可以省去安装115.28.141.137个人私服已配置好

第二步添加依赖包

<dependency>
			<groupId>org.springframework.amqp</groupId>
			<artifactId>spring-rabbit</artifactId>
			<version>1.3.5.RELEASE</version>
		</dependency>
	
		<dependency>
			<groupId>commons-lang</groupId>
			<artifactId>commons-lang</artifactId>
			<version>2.6</version>
		</dependency>

第三步配置生产者工程服务配置

<?xml version="1.0" encoding="UTF-8"?>
<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:task="http://www.springframework.org/schema/task"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
      http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
      http://www.springframework.org/schema/context
      http://www.springframework.org/schema/context/spring-context.xsd
      http://www.springframework.org/schema/integration/ip 
      http://www.springframework.org/schema/integration/ip/spring-integration-ip-2.0.xsd
      http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
	  http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd
	  http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
	  http://www.springframework.org/schema/rabbit
      http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd
      ">

	<!-- 引入jdbc配置文件 -->
	<context:property-placeholder location="classpath:conf/jdbc.properties" />

	<!-- 扫描文件(自动将servicec层注入) -->
	<context:component-scan base-package="mq.service" />
	<!-- 连接服务配置 -->
	<rabbit:connection-factory id="connectionFactory"
		host="115.28.141.137" 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.util.FastJsonMessageConverter"></bean>

	<!-- spring template声明-->
	<rabbit:template exchange="my-mq-exchange" id="amqpTemplate"
		connection-factory="connectionFactory" message-converter="jsonMessageConverter" />
</beans>

FastJsonMessageConverter 类代码如下

package mq.util;

import java.io.UnsupportedEncodingException;

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 com.alibaba.druid.support.logging.Log;
import com.alibaba.druid.support.logging.LogFactory;
import com.alibaba.fastjson.JSON;

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;
    }
     
    @SuppressWarnings("unchecked")
	public <T> T fromMessage(Message message,T t) {
        String json = "";
        try {
            json = new String(message.getBody(),"UTF-8");
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        return (T) JSON.parseObject(json, t.getClass());
    }   
     
 
    protected Message createMessage(Object objectToConvert,
            MessageProperties messageProperties)
            throws MessageConversionException {
        byte[] bytes = null;
        try {
            String jsonString = JSON.toJSONString(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);
 
    }
}

生产者代码Producer

package mq.service;



import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
 
@Service
public class ProducerMq {
     
    @Autowired
    private AmqpTemplate amqpTemplate;
     
    public void sendDataToCrQueue(Object obj) {
        amqpTemplate.convertAndSend("queue_one_key", obj);
    }   
}

第四步配置消费者服务工程配置

<?xml version="1.0" encoding="UTF-8"?>
<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:task="http://www.springframework.org/schema/task"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
      http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
      http://www.springframework.org/schema/context
      http://www.springframework.org/schema/context/spring-context.xsd
      http://www.springframework.org/schema/integration/ip 
      http://www.springframework.org/schema/integration/ip/spring-integration-ip-2.0.xsd
      http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
	  http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd
	  http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
	  http://www.springframework.org/schema/rabbit
      http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd
      ">

	<!-- 引入jdbc配置文件 -->
	<context:property-placeholder location="classpath:conf/jdbc.properties" />

	<!-- 扫描文件(自动将servicec层注入) -->
	<context:component-scan base-package="mq.service" />
	<task:executor id="taskExecutor" pool-size="1-4" queue-capacity="128" />  
	 <!-- 连接服务配置  -->
   <rabbit:connection-factory id="connectionFactory" host="115.28.141.137" 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>
    <bean id="queueOneLitener" class="mq.service.QueueOneLitener" />
        <!-- queue litener  观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象-->
    <rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto" task-executor="taskExecutor">
        <rabbit:listener queues="queue_one" ref="queueOneLitener"/>  </rabbit:listener-container>
</beans>

监听器代码queueOneLitener

package mq.service;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
 
public class QueueOneLitener implements  MessageListener{
    public void onMessage(Message message) {
    
        System.out.println(" data :" +new String(message.getBody()));
    }
}

第五步测试

消费者端控制层生产数据

控制层Controller

package mq.controller;

import javax.annotation.Resource;

import mq.service.ProducerMq;

import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;

@Controller
public class MessageController {
    @Resource
    private ProducerMq producer;
	@RequestMapping("/producer")
	public String producer() throws Exception {
		for(int i=0;i<100;i++){
			producer.sendDataToCrQueue("data"+i);
		}
		return "index";
	}


}

运行结果

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏LanceToBigData

JavaWeb(三)JSP概述

一、JSP概述 1.1、JSP简介   一种动态网页开发技术。它使用JSP标签在HTML网页中插入Java代码。标签通常以<%开头以%>结束。JSP是一种Jav...

39160
来自专栏java学习

使用intellij idea搭建MAVEN+SSM(Spring+SpringMVC+MyBatis)框架

Spring是一个开源框架,Spring是于2003 年兴起的一个轻量级的Java 开发框架,由Rod Johnson 在其著作Expert One-On-On...

50950
来自专栏Java Web

使用RESTful风格开发Java Web

REST是REpresentational State Transfer的缩写(一般中文翻译为表述性状态转移),REST 是一种体系结构,而 HTTP 是一种包...

23250
来自专栏三丰SanFeng

无锁编程(六) - seqlock(顺序锁)

seqlock(顺序锁) 用于能够区分读与写的场合,并且是读操作很多、写操作很少,写操作的优先权大于读操作。 seqlock的实现思路是,用一个递增的整型数表...

24770
来自专栏noteless

eclipse 创建maven 项目 动态web工程完整示例 maven 整合springmvc整合mybatis

如果搜索不到内容,或者有报错信息(index downloads are disabled ,search results may be incomplete)

20720
来自专栏云原生架构实践

Jhipster技术栈定制 - 基于UAA的微服务之间安全调用

3个微服务都是通过Jhipster生成。 工程代码生成完之后,根据上一节启动的组件的实际情况,修改微服务配置文件中Eureka和database相关的配置。

1K30
来自专栏向治洪

Android NDk环境配置

概论 NDK全称是Native Development Kit,NDK提供了一系列的工具,帮助开发者快速开发C(或C++)的动态库,并能自动将so和java应...

207100
来自专栏JavaWeb

项目中Spring 声明式事务使用的一些坑点分析01

81080
来自专栏帘卷西风的专栏

Cmake在windows支持预编译头文件(stdafx.h)

      最近一直在研究cmake构建项目,之前接触cmake的时候就感觉不太喜欢cmake,觉得它太乱了,产生了太多的中间文件,产生的项目文件也不是特别友好...

14920
来自专栏流柯技术学院

linux下安装rzsz

wget http://freeware.sgi.com/source/rzsz/rzsz-3.48.tar.gz

62310

扫码关注云+社区

领取腾讯云代金券