rocketmq的安装
rocketMq官网:http://rocketmq.apache.org/
下载比较简单.rocketmq-3.5.8.zip,和rocketmq-externals-master扩展的,里面有控制台,比较方便也下载下来
我以前下载过,就不过多介绍,解压出来:
这是rocketmq源码,
这是扩展源码:
因为是源码都直接运行不了的,需要进行编译,因为是java语言编写的,肯定需要java环境,直接执行install.bat,然后进入target目录.rocket就是一个maven项目.
targe里面就可以直接执行了
执行文件都在bin里面:
分别执行这两个,一般没报错就成功了,
console的页面控制台基本同理:
有个window的批处理文件,直接点击很方便很人性化.然后也是一个springboot项目,看到最后的启动成功就ok了,看得到端口,12581
刚进去的画面很酷,可以更换语言,简体中文.很方便,都是中文很方便
启动注意我使用的老版本,新版本好像不用设置了:
先启动mynameserver
在启动mqbroker的时候需要指定autoCreateTopicEnable=true。例如：
nohup sh mqbroker -n 192.168.180.133:9876 autoCreateTopicEnable=true > ~/logs/rocketmqlogs/broker.log 2>&1 &
window系统下：
在window系统下需要在cmd中启动mqbroker才行。命令格式如下：
mqbroker.exe -n localhost:9876 autoCreateTopicEnable=true
springboot整合rocketmq
网上很多有整合rocketmq的,但是部分没有使用官方rocketmq-spring-boot-starter,因为官方start是近几年出的,使用起来很简单,我这里就做一个简单的demo好了…
rocketmq-spring-boot-starter官方文档:
https://github.com/apache/rocketmq-spring/blob/master/README_zh_CN.md
我自己整合过部分springboot,然后这整合我就放一起了,因为是简单的整合使用.所以,不要说我太菜,尽管说了我也不听,/手动狗头
首先引入依赖:
版本可以改的,当然也最好提到properties里面,这里的话就可以直接写里面.
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.0.3</version>
</dependency>
测试的话,我就将topic都直接写死了,一般都会写在配置文件里面的…公司中使用需要注意的.在配置文件里面加上:
基本的配置
rocketmq:
name-server: 127.0.0.1:9876
producer:
group: my-group
编写个生产者
测试类,一个是发送简单的String,一个是发送类:
import com.mr.pojo.OrderPaidEvent;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.context.junit4.SpringRunner;
import javax.annotation.Resource;
import java.math.BigDecimal;
/**
* @ClassName RocketMqTest
* @Description RocketMqTest
* @Author stack
* @Version 1.0
* @since 2019/6/23 18:32
*/
@RunWith(SpringRunner.class)
@SpringBootTest
public class RocketMqTest {
@Resource
private RocketMQTemplate rocketMQTemplate;
@Test
public void testRocketMq1() {
String name = "aaa";
rocketMQTemplate.convertAndSend("test-topic-1", name);
rocketMQTemplate.send("test-topic-2", MessageBuilder.withPayload(new OrderPaidEvent("aa,22",new BigDecimal("22"))).build());
System.err.println("发送成功...");
}
}
实体类:
@Data
@AllArgsConstructor
public class OrderPaidEvent implements Serializable {
private String orderId;
private BigDecimal paidMoney;
}
编写个消费者
@Service
@RocketMQMessageListener(topic = "test-topic-2", consumerGroup = "my-consumer_test-topic-2")
public class OrderPaidEventConsumer implements RocketMQListener<OrderPaidEvent> {
@Override
public void onMessage(OrderPaidEvent message) {
System.out.print("------- OrderPaidEventConsumer received:"+ JSON.toJSONString(message));
}
}
测试
然后测试一波:先启动消费者:
在测试类发送两个消息
然后接受的时候一直报错：
后面请教了室友邱大神，然后他定位到，是jackjson解析的时候出现了问题，一步一步定位，发现手写的getset方法没问题，只要一加上@AllArgsConstructor注解就报错。。
然后将实体类改为:
public class OrderPaidEvent implements Serializable {
private String orderId;
private int paidMoney;
public OrderPaidEvent() {
}
public OrderPaidEvent(String orderId, int paidMoney) {
this.orderId = orderId;
this.paidMoney = paidMoney;
}
public String getOrderId() {
return orderId;
}
public void setOrderId(String orderId) {
this.orderId = orderId;
}
public int getPaidMoney() {
return paidMoney;
}
public void setPaidMoney(int paidMoney) {
this.paidMoney = paidMoney;
}
}
然后重启，发现问题解决，佩服邱大神。。厉害了
然后在控制台也可以看的见：
个人源码
https://github.com/stackXu/springboot-study
参考
官方:https://github.com/apache/rocketmq-spring/blob/master/README_zh_CN.md