spring整合rocketMq使用（完整代码，适合小白）

pom.xml：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<groupId>com.th</groupId>
	<artifactId>rocketMq</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<packaging>war</packaging>

	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<junit.version>4.12</junit.version>
		<spring.version>4.2.0.RELEASE</spring.version>
		<httpclient.version>4.3.1</httpclient.version>
		<jedis.version>2.7.2</jedis.version>
	</properties>

	<dependencies>
		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-test</artifactId>
			<version>${spring.version}</version>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-webmvc</artifactId>
			<version>${spring.version}</version>
		</dependency>
		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-core</artifactId>
			<version>${spring.version}</version>
		</dependency>

		<dependency>
			<groupId>com.alibaba.rocketmq</groupId>
			<artifactId>rocketmq-client</artifactId>
			<version>3.2.6</version>
		</dependency>
		<dependency>
			<groupId>com.alibaba.rocketmq</groupId>
			<artifactId>rocketmq-all</artifactId>
			<version>3.2.6</version>
			<type>pom</type>
		</dependency>

		<!-- https://mvnrepository.com/artifact/junit/junit -->
		<dependency>
			<groupId>junit</groupId>
			<artifactId>junit</artifactId>
			<version>${junit.version}</version>
			<scope>test</scope>
		</dependency>

		<dependency>
			<groupId>javax.servlet</groupId>
			<artifactId>jstl</artifactId>
			<version>1.2</version>
		</dependency>


	</dependencies>

</project>

springApplicationContext配置：

<?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:mvc="http://www.springframework.org/schema/mvc"
	xsi:schemaLocation="http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-4.2.xsd
		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-3.0.xsd">

	<context:property-placeholder location="classpath:config.properties" />

	<!-- 注解解析 -->
	<context:annotation-config />

	<context:component-scan base-package="com.th" />

	<mvc:annotation-driven />

	<bean class="org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerMapping" />

	<!-- rocketmq配置 -->
	<bean id="messageListeners" class="com.th.listener.MessageListenerImpl"></bean>
	<!-- 导入Spring配置文件 -->
	<bean id="rocketmqConsumer" class="com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer" init-method="start" destroy-method="shutdown">
		<property name="consumerGroup" value="${rocketmq.consumerGroup}" />
		<property name="namesrvAddr" value="${rocketmq.namesrvAddr}" />
		<property name="messageListener" ref="messageListeners" />
		<property name="subscription">
			<map>
				<entry key="${rocketmq.topic}" value="${rocketmq.tags}" />
			</map>
		</property>
	</bean>


	<bean id="rocketMQProducer" class="com.th.producer.RocketMQProducer" init-method="init" destroy-method="destroy">
		<property name="producerGroup" value="${rocketmq.producer.group}" />
		<property name="namesrvAddr" value="${rocketmq.namesrvAddr}" />
		<!-- 失败重试次数 <property name="retryTimes" value="${rocketmq.producer.retryTimes}" /> -->
	</bean>


	<bean id="viewResolver" class="org.springframework.web.servlet.view.InternalResourceViewResolver">
		<property name="prefix" value="/WEB-INF/pages/" />
		<property name="suffix" value=".jsp"></property>
	</bean>


</beans>

config.properties:

rocketmq.consumerGroup=thConsumerGroup
rocketmq.namesrvAddr=192.168.0.117:9876;192.168.0.118:9876
rocketmq.tags=TAG1
rocketmq.producer.group=thProducerGroup
rocketmq.topic=TestTopic1

UserController:

package com.th.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.th.producer.RocketMQProducer;

@Controller
public class UserController {
	@Autowired
	@Qualifier("rocketMQProducer")
	private RocketMQProducer producer;

	@RequestMapping(value = "/test", method = RequestMethod.GET)
	public String test() throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
		for (int i = 0; i < 100; i++) {
			Message msg = new Message("TestTopic1", "TAG1", (i + "這是spring集成").getBytes());
			SendResult result = producer.getDefaultMQProducer().send(msg);
			System.out.println(result);
			System.out.println(1);
		}
		return "hello";
	}
}

监听类（消费者）：

package com.th.listener;

import java.io.UnsupportedEncodingException;
import java.util.List;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;

public class MessageListenerImpl implements MessageListenerConcurrently {

	@Override
	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
		for (MessageExt msg : msgs) {
			try {
				System.out.println(">>>>" + new String(msg.getBody(), "UTF-8"));
			} catch (UnsupportedEncodingException e) {
				e.printStackTrace();
				return ConsumeConcurrentlyStatus.RECONSUME_LATER;
			}
		}
		// 若是没有异常会认为都成功消费
		return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
	}
}

生成者：

package com.th.producer;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;

public class RocketMQProducer {
	// private static final Logger logger =
	// LoggerFactory.getLogger(RocketMQProducer.class);

	private DefaultMQProducer defaultMQProducer;
	private String producerGroup;
	private String namesrvAddr;
	//private String instanceName;
	//private int retryTimes;

	public void init() throws MQClientException {
		this.defaultMQProducer = new DefaultMQProducer(this.producerGroup);
		defaultMQProducer.setNamesrvAddr(this.namesrvAddr);
		//defaultMQProducer.setInstanceName(this.instanceName);
		//defaultMQProducer.setRetryTimesWhenSendFailed(this.retryTimes);
		defaultMQProducer.start();
		// logger.info("rocketMQ初始化生产者完成[producerGroup：" + producerGroup +
		// "，instanceName：" + instanceName + "]");
	}

	public void destroy() {
		defaultMQProducer.shutdown();
		// logger.info("rocketMQ生产者[producerGroup: " + producerGroup +
		// ",instanceName: " + instanceName + "]已中止");
	}

	public DefaultMQProducer getDefaultMQProducer() {
		return defaultMQProducer;
	}

	public void setProducerGroup(String producerGroup) {
		this.producerGroup = producerGroup;
	}

	public void setNamesrvAddr(String namesrvAddr) {
		this.namesrvAddr = namesrvAddr;
	}

	public void setInstanceName(String instanceName) {
		//this.instanceName = instanceName;
	}

	public void setRetryTimes(int retryTimes) {
		//this.retryTimes = retryTimes;
	}
}

web.xml:

<!DOCTYPE web-app PUBLIC
 "-//Sun Microsystems, Inc.//DTD Web Application 2.3//EN"
 "http://java.sun.com/dtd/web-app_2_3.dtd" >

<web-app>
	<display-name>Archetype Created Web Application</display-name>

	<listener>
		<listener-class>org.springframework.web.context.request.RequestContextListener</listener-class>
	</listener>

	<filter>
		<filter-name>CharacterEncodingFilter</filter-name>
		<filter-class>org.springframework.web.filter.CharacterEncodingFilter</filter-class>
		<init-param>
			<param-name>encoding</param-name>
			<param-value>UTF-8</param-value>
		</init-param>
		<init-param>
			<param-name>forceEncoding</param-name>
			<param-value>true</param-value>
		</init-param>
	</filter>
	<filter-mapping>
		<filter-name>CharacterEncodingFilter</filter-name>
		<url-pattern>/*</url-pattern>
	</filter-mapping>

	<servlet>
		<!-- 配置DispatcherServlet -->
		<servlet-name>springMvc</servlet-name>
		<servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
		<!-- 指定spring mvc配置文件位置 不指定使用默认状况 -->
		<init-param>
			<param-name>contextConfigLocation</param-name>
			<param-value>classpath:ApplicationContext.xml</param-value>
		</init-param>
		<!-- 设置启动顺序 -->
		<load-on-startup>1</load-on-startup>
	</servlet>

	<!-- ServLet 匹配映射 -->
	<servlet-mapping>
		<servlet-name>springMvc</servlet-name>
		<url-pattern>/</url-pattern>
	</servlet-mapping>
	
	<welcome-file-list>
		<welcome-file>index.jsp</welcome-file>
	</welcome-file-list>

</web-app>

最后启动项目：

浏览器访问地址： http://localhost:8089/rocketMq/test

查看控制台，出现信息消费则成功