当前位置:   article > 正文

SpringMVC集成Rocketmq_springmvc集成java config rocketmq

springmvc集成java config rocketmq

软件环境:
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>
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

生产者
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;
	}
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50

消费者
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;
	    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87

生产者配置文件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>
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17

消费者配置文件: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>
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17

然后将这两个xml导入到spring的配置文件里

<import resource="classpath:rocketmq/producer.xml" />
<import resource="classpath:rocketmq/consumer.xml" />
  • 1
  • 2

到此就可以使用了。

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/小小林熬夜学编程/article/detail/434677
推荐阅读
相关标签
  

闽ICP备14008679号