首页 > > 程序设计 > Java技术 >

Spring + RocketMQ使用

2019-08-16 11:13:23来源：博客园 阅读 ()

新老客户大回馈,云服务器低至5折

Spring + RocketMQ使用

本文所介绍环境为win7环境下运行， 从官方github中（https://github.com/alibaba/RocketMQ）下载RocketMQ-master.zip，版本为v3.5.8，解压并进入根目录，运行命令install.bat， 安装完成后进入目录\target\alibaba-rocketmq-broker\alibaba-rocketmq\bin，打开两个命令行窗口，分别使用以下命令启动rocketmq

启动nameserver
mqnamesrv.exe
启动broker
mqbroker -n 127.0.0.1:9876

1、编写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.hode</groupId>
    <artifactId>rocketmq</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    
    <properties>
        <spring.version>4.3.2.RELEASE</spring.version>
        <junit.version>4.12</junit.version>
        <log4j.version>1.2.17</log4j.version>
        <rocketmq.version>3.2.6</rocketmq.version>
        <slf4j.version>1.7.12</slf4j.version>
    </properties>
    
    <dependencies>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-core</artifactId>
            <version>${spring.version}</version>
        </dependency>
        
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
            <version>${spring.version}</version>
        </dependency>
        
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>${log4j.version}</version>
        </dependency>
        
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        
        <dependency>
            <groupId>com.alibaba.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>${rocketmq.version}</version>
        </dependency>
        
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>${junit.version}</version>
            <scope>test</scope>
        </dependency>
        
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-test</artifactId>
            <version>${spring.version}</version>
            <scope>test</scope>
        </dependency>
        
    </dependencies>
    
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            
        </plugins>
    </build>
    
</project>

2、编写spring配置文件applicationContext-consumer.xml,applicationContext-producer.xml以及log4j.properties，内容如下

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

    <bean id="producer" class="com.hode.rocketmq.Consumer" init-method="init" destroy-method="destroy">
        <constructor-arg name="consumerGroup" value="rocketmq-test" />
        <constructor-arg name="namesrvAddr" value="127.0.0.1:9876" />
        <constructor-arg name="instanceName" value="test" />
        <constructor-arg name="topic" value="testTopic" />
        <constructor-arg name="messageListener" ref="messageListener" />
    </bean>
    
    <bean id="messageListener" class="com.hode.rocketmq.StringMessageListener" />
    
</beans>

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

    <bean id="producer" class="com.hode.rocketmq.Producer" init-method="init" destroy-method="destroy">
        <constructor-arg name="producerGroup" value="rocketmq-test" />
        <constructor-arg name="namesrvAddr" value="127.0.0.1:9876" />
        <constructor-arg name="instanceName" value="test" />
    </bean>
    
</beans>

log4j.properties

log4j.rootLogger=INFO,Console
log4j.appender.Console=org.apache.log4j.ConsoleAppender
log4j.appender.Console.layout=org.apache.log4j.PatternLayout
log4j.appender.Console.layout.ConversionPattern=%-4r %d{yyyy-MM-dd HH:mm:ss,SSS} [%t] %-5p %c %x - %m%n

log4j.logger.com.hode=DEBUG

3、编写类StringMessageListener.java,Producer.java,Consumer.java

StringMessageListener.java

import java.util.List;

import org.apache.log4j.Logger;

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 StringMessageListener implements MessageListenerConcurrently{

    private Logger log = Logger.getLogger(getClass());
    
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {
        for (MessageExt msg : msgs) {
            log.info("msg : " + new String(msg.getBody()));
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

}

Producer.java

import org.apache.log4j.Logger;

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

/**
 * 生产
 */
public class Producer {
    
    protected Logger log = Logger.getLogger(getClass());
    
    private String producerGroup;
    
    private String namesrvAddr;
    
    private String instanceName;
    
    private DefaultMQProducer producer;
    
    public DefaultMQProducer getProducer() {
        return producer;
    }

    public Producer(String producerGroup,String namesrvAddr,String instanceName){
        this.producerGroup = producerGroup;
        this.namesrvAddr = namesrvAddr;
        this.instanceName = instanceName;
    }
    
    public void init() throws MQClientException{
        log.info("start init DefaultMQProducer...");
        producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(namesrvAddr);
        producer.setInstanceName(instanceName);
        producer.start();
        log.info("DefaultMQProducer init success.");
    }
    
    public void destroy(){
        log.info("start destroy DefaultMQProducer...");
        producer.shutdown();
        log.info("DefaultMQProducer destroy success.");
    }

}

Consumer.java

import org.apache.log4j.Logger;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;

public class Consumer {

    private Logger log = Logger.getLogger(getClass());
    
    private DefaultMQPushConsumer consumer;
    
    private String consumerGroup;
    
    private String namesrvAddr;
    
    private String instanceName;
    
    private String topic;
    
    private MessageListenerConcurrently messageListener;
    
    public Consumer(String consumerGroup,String namesrvAddr,String instanceName,String topic,MessageListenerConcurrently messageListener){
        this.consumerGroup = consumerGroup;
        this.namesrvAddr = namesrvAddr;
        this.instanceName = instanceName;
        this.topic = topic;
        this.messageListener = messageListener;
    }
    
    public void init() throws Exception{
        log.info("start init DefaultMQPushConsumer...");
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); //从队列头部开始消费
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setInstanceName(instanceName);
        consumer.subscribe(topic, "*");
        consumer.registerMessageListener(messageListener);
        consumer.start();
        log.info("DefaultMQPushConsumer init ok.");
    }

    public void destroy(){
        log.info("start destroy DefaultMQPushConsumer...");
        consumer.shutdown();
        log.info("DefaultMQPushConsumer destroy success.");
    }
    
    public DefaultMQPushConsumer getConsumer() {
        return consumer;
    }

}

4、编写测试类

ProducerTest.java

import java.util.Date;

import org.apache.log4j.Logger;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendCallback;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;

public class ProducerTest {

    private static Logger log = Logger.getLogger(ProducerTest.class);
    
    private static ApplicationContext context;
    
    public static void main(String[] args) throws Exception{
        context = new ClassPathXmlApplicationContext("classpath:applicationContext-producer.xml");
        Producer producer = context.getBean(Producer.class);
        DefaultMQProducer p = producer.getProducer();
        
        String message = "test messgae"+new Date();
        Message msg = new Message("testTopic",message.getBytes());
        log.info(message);
        p.send(msg, new SendCallback(){

            @Override
            public void onSuccess(SendResult sendResult) {
                log.info(sendResult.getSendStatus().name());
                log.info("onSuccess");
                
                producer.destroy();
            }

            @Override
            public void onException(Throwable e) {
                log.error("onException");
            }
        });
        
    }
    
}

ConsumerTest.java

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class ConsumerTest {

    private static ApplicationContext context;
        
    public static void main(String[] args) throws Exception{
        context = new ClassPathXmlApplicationContext("classpath:applicationContext-consumer.xml");
        Consumer consumer = context.getBean(Consumer.class);
        
        Thread.sleep(20*1000);
        
        System.out.println("end");
        consumer.destroy();
    }
    
}

分别运行生产端及消费端完成测试，结束。


原文链接:https://www.cnblogs.com/alterem/p/11249001.html
如有疑问请与原作者联系

标签：

版权申明：本站文章部分自网络，如有侵权，请联系：west999com@outlook.com
特别注意：本站所有转载文章言论不代表本站观点，本站所提供的摄影照片，插画，设计作品，如需使用，请与原作者联系，版权归原作者所有

上一篇：详述Spring对数据校验支持的核心API：SmartValidator

下一篇：对比分析HashMap、LinkedHashMap、TreeMap

相关文章

IDC资讯： 主机资讯 注册资讯 托管资讯 vps资讯 网站建设

网站运营： 建站经验 策划盈利 搜索优化 网站推广 免费资源

网站联盟： 联盟新闻 联盟介绍 联盟点评 网赚技巧

行业资讯： 搜索引擎 网络游戏 电子商务 广告传媒

网络编程： Asp.Net编程 Asp编程 Php编程 Xml编程 Access Mssql Mysql 其它

服务器技术： Web服务器 Ftp服务器 Mail服务器 Dns服务器 安全防护

软件技巧： 其它软件 Word Excel Powerpoint Ghost Vista QQ空间 QQ FlashGet 迅雷

网页制作： FrontPages Dreamweaver Javascript css photoshop fireworks Flash

程序设计： Java技术 C/C++ VB delphi

网络知识： 网络协议 网络安全 网络管理 组网方案 Cisco技术

操作系统： Win2000 WinXP Win2003 Mac OS Linux FreeBSD

热门词条
最新资讯
热门关注
热门标签