Linux下RocketMQ下载安装教程以及RocketMQ使用教程和代码
一、安装及启动
1、将zip文件上传到linux的~下,然后解压:
执行:unzip rocketmq-all-4.4.0-bin-release.zip,若unzip命令不存在,执行:yum install -y unzip zip
2、移动并重命名
执行:mv rocketmq-all-4.4.0-bin-release /usr/local/rocketmq
3、进入/usr/local/rocketmq/bin目录下
执行:cd /usr/local/rocketmq/bin
4、如果是在真实的linux服务器上安装rocketmq,则内存是足够的,如果是在虚拟机上安装rocketmq进行学习用,可能虚拟机的内存只分配了1G,那么是不够用的。启动rocketmq可能会启动不成功。此时就先修改一下启动所需的内存参数。
(1)修改runserver.sh
执行:vi runserver.sh
将其内存修改为128m并保存,如下:
(2)修改runbroker.sh
同理将其修改为128m(与修改runserver.sh一样)
5、启动nameserver,在bin目录下执行:nohup sh mqnamesrv &,查看启动是否成功的日志命令:tail -f ~/logs/rocketmqlogs/namesrv.log ,打印如下表示启动成功:
6、启动broker,在bin目录下执行:nohup sh mqbroker -n localhost:9876 & ,也可以 nohup sh mqbroker -c ../conf/broker.conf -n 192.168.153.128:9876 autoCreateTopicEnable=true &
这样启动的服务器可以自动创建主题(客户端),不过生产一般不推荐.
查看日志命令:tail -f ~/logs/rocketmqlogs/broker.log,打印如下表示启动成功:
三、安装可视化插件
1、下载地址:https://pan.baidu.com/s/16ZoOkYEBtNq1sGp0cuk3cA 提取码:ms2g
2、上传到linux自己指定一个目录并使用unzip命令解压(步骤略,可参考上面rocketmq的上传解压步骤)
3、进入rocketmq-console/src/main/resources目录下,打开application.properties并修改保存如下:
4、进入‘\rocketmq-externals\rocketmq-console’文件夹,执行‘mvn clean package -Dmaven.test.skip=true’,编译生成,该过程耗时较长,最终生成成功后如下:
5、便已完成,会生成target目录,进入target目录后执行:java -jar rocketmq-console-ng-1.0.1.jar,启动完成后,
在浏览器输入:http://192.168.153.128:9876,出现如下界面
完毕!
生产者代码
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" />
到此就可以使用了。
rocketMq发送端代码:
package com.wechat.tool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
public class RocketMqProducer {
private final Logger logger = LoggerFactory.getLogger(RocketMqProducer.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();
logger.info("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;
}
}
该对象的实例初始化与销毁都是由spring来控制,RocketMqProducer需要保持单例,主要是由spring配置文件控制,下面是RocketMqProducer相关配置
<bean id="rocketMqProducer" class="com.wechat.tool.RocketMqProducer"
init-method="init" destroy-method="destroy" scope="singleton">
<property name="producerGroup" value="PushConsumer" />
<property name="namesrvAddr" value="127.0.0.1:9876" />
</bean>
springmvc控制层这块主要调用RocketMqProducer单例对象,对RocketMq进行消息队列的发送,下列只列出部分控制层代码
@Autowired
private RocketMqProducer rocketMqProducer;
@RequestMapping(value = "/coupon/couponsend.htm")
@ResponseBody()
public Map<String,String> couponsend(){
String sendStatus ="";
String msgId ="";
try{
String msgtext="优惠券领取"+DTUtil.getTodayString4();
Message msg = new Message("PushTopic",
"push",
"coupon"+DTUtil.getNowLongTime(),
msgtext.getBytes());
SendResult result = rocketMqProducer.getDefaultMQProducer().send(msg);
System.out.println("id:" + result.getMsgId() +
" result:" + result.getSendStatus());
msgId = result.getMsgId();
sendStatus=result.getSendStatus()+"";
}catch (Exception e) {
e.printStackTrace();
}
Map<String,String> m=new HashMap<String, String>();
m.put("sendStatus", sendStatus);
m.put("msgId", msgId);
return m;
}
rocketMq消费者端:
package com.wechat.test;
import java.util.List;
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.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageExt;
public class PushConsumer {
public static void main(String[] args){
DefaultMQPushConsumer consumer =
new DefaultMQPushConsumer("PushConsumer");
consumer.setNamesrvAddr("127.0.0.1:9876");
try {
//订阅PushTopic下Tag为push的消息
consumer.subscribe("PushTopic", "push");
//程序第一次启动从消息队列头取数据
consumer.setConsumeFromWhere(
ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.registerMessageListener(
new MessageListenerConcurrently() {
public ConsumeConcurrentlyStatus consumeMessage(
List<MessageExt> list,
ConsumeConcurrentlyContext Context) {
Message msg = list.get(0);
System.out.println(msg.getKeys());
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}
);
consumer.start();
} catch (Exception e) {
e.printStackTrace();
}
}
}
消费者开启后就不用关闭