SpringMVC集成Rocketmq

ゞ浴缸里的玫瑰 2022-05-17 04:15 296阅读 0赞

软件环境：  
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的配置文件里

到此就可以使用了。