赞
踩
软件环境:
windows7
Rocketmq 4.2
(上):安装配置RocketMq
一:配置环境变量
NAMESRV_ADDR=127.0.0.1:9876和rocketMQ的安装目录ROCKETMQ_HOME=
D:\rocketmq-4.2.0
二:启动nameserver
在bin目录下,使用mqnamesrv.cmd
启动成功输出:The Name Server boot success.
三:启动broker
使用命令mqbroker.cmd
启动broker时报错:
错误: 找不到或无法加载主类 Files\Java\jdk1.8.0_101\lib;C:\Program
解决办法:
runbroker.cmd里修改
set "JAVA_OPT=%JAVA_OPT% -cp %CLASSPATH%"这行为:set “JAVA_OPT=%JAVA_OPT% -cp “%CLASSPATH%””
(下)springMVC集成rocketMQ
引入依赖:
<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>
生产者
MyProducer.java
import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.rocketmq.client.exception.MQClientException; import com.alibaba.rocketmq.client.producer.DefaultMQProducer; public class MyProducer { private final Logger logger = LoggerFactory.getLogger(MyProducer.class); private DefaultMQProducer defaultMQProducer; private String producerGroup; private String namesrvAddr; /** * Spring bean init-method */ public void init() throws MQClientException { // 参数信息 logger.info("DefaultMQProducer initialize!"); logger.info(producerGroup); logger.info(namesrvAddr); // 初始化 defaultMQProducer = new DefaultMQProducer(producerGroup); defaultMQProducer.setNamesrvAddr(namesrvAddr); defaultMQProducer.setInstanceName(String.valueOf(System .currentTimeMillis())); defaultMQProducer.start(); System.out.println("DefaultMQProudcer start success!"); } /** * Spring bean destroy-method */ public void destroy() { defaultMQProducer.shutdown(); } public DefaultMQProducer getDefaultMQProducer() { return defaultMQProducer; } // ---------------setter ----------------- public void setProducerGroup(String producerGroup) { this.producerGroup = producerGroup; } public void setNamesrvAddr(String namesrvAddr) { this.namesrvAddr = namesrvAddr; } }
消费者
MyConsumer .java
import java.util.List; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer; 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.client.exception.MQClientException; import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere; import com.alibaba.rocketmq.common.message.MessageExt; import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel; public class MyConsumer { private final Logger logger = LoggerFactory.getLogger(MyConsumer.class); private DefaultMQPushConsumer defaultMQPushConsumer; private String namesrvAddr; private String consumerGroup; /** * Spring bean init-method */ public void init() throws InterruptedException, MQClientException { // 参数信息 logger.debug("DefaultMQPushConsumer initialize!"); logger.debug(consumerGroup); logger.debug(namesrvAddr); // 一个应用创建一个Consumer,由应用来维护此对象,可以设置为全局对象或者单例<br> // 注意:ConsumerGroupName需要由应用来保证唯一 defaultMQPushConsumer = new DefaultMQPushConsumer(consumerGroup); defaultMQPushConsumer.setNamesrvAddr(namesrvAddr); defaultMQPushConsumer.setInstanceName(String.valueOf(System.currentTimeMillis())); // 订阅指定MyTopic下tags等于MyTag defaultMQPushConsumer.subscribe("MyTopic", "MyTag"); // 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br> // 如果非第一次启动,那么按照上次消费的位置继续消费 defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); // 设置为集群消费(区别于广播消费) defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING); defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() { // 默认msgs里只有一条消息,可以通过设置consumeMessageBatchMaxSize参数来批量接收消息 @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { MessageExt msg = msgs.get(0); if (msg.getTopic().equals("MyTopic")) { // TODO 执行Topic的消费逻辑 System.out.println("接收到了消息啊=========="+msg.toString()); if (msg.getTags() != null && msg.getTags().equals("MyTag")) { // TODO 执行Tag的消费 } } // 如果没有return success ,consumer会重新消费该消息,直到return success return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); // Consumer对象在使用之前必须要调用start初始化,初始化一次即可<br> defaultMQPushConsumer.start(); logger.debug("DefaultMQPushConsumer start success!"); } /** * Spring bean destroy-method */ public void destroy() { defaultMQPushConsumer.shutdown(); } // ----------------- setter -------------------- public void setNamesrvAddr(String namesrvAddr) { this.namesrvAddr = namesrvAddr; } public void setConsumerGroup(String consumerGroup) { this.consumerGroup = consumerGroup; } }
生产者配置文件producer.xml
<?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" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-3.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd"> <bean id="myProducer" class="net.sinolbs.rocketmq.MyProducer" init-method="init" destroy-method="destroy" scope="singleton"> <property name="producerGroup" value="MyProducerGroup" /> <property name="namesrvAddr" value="127.0.0.1:9876" /> </bean> </beans>
消费者配置文件:consumer.xml
<?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" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-3.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd"> <bean class="net.sinolbs.test.rocketmq.MyConsumer" init-method="init" destroy-method="destroy" scope="singleton"> <property name="consumerGroup" value="MyConsumerGroup" /> <property name="namesrvAddr" value="127.0.0.1:9876" /> </bean> </beans>
然后将这两个xml导入到spring的配置文件里
<import resource="classpath:rocketmq/producer.xml" />
<import resource="classpath:rocketmq/consumer.xml" />
到此就可以使用了。
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。