1.从官网下载对应的二进制文件，直接解压即可使用
对应官网下载地址为 ：Downloading the Apache RocketMQ Releases - Apache RocketMQhttps://rocketmq.apache.org/dowloading/releases/
2.启动(window中启动命令)
1）启动namesrv
.binmqnamesrv.cmd
2) 启动broker
.binmqbroker.cmd -n localhost:9876 autoCreateTopicEnable=true
对于broker启动不成功，可以尝试以下方法
C:/Users/用户名/store，将这个目录和目录下所有文件全部删除，然后再输入broker的启动命令
3.pom文件引入依赖
org.apache.rocketmq rocketmq-client4.9.1
4.生产者和消费者相关代码
rocketmq.name.server.address=127.0.0.1:9876
public class RocketMQConstant {
public static final String GROUP_MOMENTS = "GroupMoments" ;
public static final String TOPIC_MOMENTS = "TopicMoments" ;
}
对应业务的生产者，和处理实际业务的消费者
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.example.bilili.constant.RocketMQConstant;
import org.example.bilili.domain.Goods;
import org.example.bilili.websocket.WebSocketService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.util.StringUtils;
import javax.annotation.Resource;
import javax.websocket.Session;
import java.util.List;
@Configuration
public class RocketMQConfig {
private static final Logger log = LoggerFactory.getLogger(RocketMQConfig.class);
@Value("${rocketmq.name.server.address}")
private String nameServerAddr;
@Resource(name = "stringRedisTemplate")
private StringRedisTemplate redisTemplate ;
@Bean("momentsProducer")
public DefaultMQProducer momentsProducer() throws Exception {
DefaultMQProducer producer = new DefaultMQProducer(RocketMQConstant.GROUP_MOMENTS);
producer.setNamesrvAddr(nameServerAddr);
producer.start();
return producer;
}
@Bean("momentsConsumer")
public DefaultMQPushConsumer momentsConsumer() throws Exception {
log.info("in momentsConsumer....");
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMQConstant.GROUP_MOMENTS);
consumer.setNamesrvAddr(nameServerAddr);
// consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe(RocketMQConstant.TOPIC_MOMENTS,"*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
log.info("consumeMessage....:{}",list.size());
for (MessageExt messageExt : list) {
String bodyStr = new String(messageExt.getBody());
Goods goods = JSONObject.parseObject(bodyStr, Goods.class);
Integer goodsId = goods.getId();
String key = "subscribed-" + String.valueOf(goodsId);
String goodsJsonStr = JSON.toJSONString(goods) ;
log.info("goods....:{}",goodsJsonStr);
String subscribedListStr = redisTemplate.opsForValue().get(key);
if(StringUtils.isEmpty(subscribedListStr)){
redisTemplate.opsForValue().set(key, goodsJsonStr);
}
}
// for(MessageExt msg:list){
// log.info("MessageExt:{}", JSONObject.toJSONString(msg));
// }
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
return consumer;
}
}
同步发送和异步发送
public class RocketMQUtil {
private static final Logger log = LoggerFactory.getLogger(RocketMQUtil.class);
public static void syncSendMsg(DefaultMQProducer producer, Message msg) throws Exception{
SendResult result = producer.send(msg);
log.info("sync send msg:{}", JSON.toJSONString(result));
}
public static void asyncSendMsgReal(DefaultMQProducer producer, Message msg) throws Exception{
producer.send(msg, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
log.info("send rocket mq success:{}",sendResult.getMsgId());
}
@Override
public void onException(Throwable throwable) {
log.info("send rocket mq exception:{}",throwable.getMessage(),throwable);
}
});
}
public static void asyncSendMsg(DefaultMQProducer producer, Message msg) throws Exception{
int messageCount = 2 ;//定义发送次数
CountDownLatch2 countDownLatch2 = new CountDownLatch2(messageCount);
for(int i=0;i
测试发送
import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.example.bilili.constant.RocketMQConstant;
import org.example.bilili.domain.Goods;
import org.example.bilili.util.RocketMQUtil;
import org.example.bilili.vo.CommonResponse;
import org.springframework.context.ApplicationContext;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.util.Date;
@RestController
@RequestMapping("/rocket-mq")
public class RocketMQController {
@Resource
private ApplicationContext applicationContext ;
@GetMapping("/send-sync-mq")
public CommonResponse sendMQTest() throws Exception{
Goods goods = new Goods();
goods.setId(1001);
goods.setCreateTime(new Date());
goods.setUpdateTime(new Date());
goods.setGoodsName("苹果11");
goods.setPrice(new BigDecimal("4999.9"));
goods.setSupply(100);
DefaultMQProducer defaultMQProducer = (DefaultMQProducer)applicationContext.getBean("momentsProducer");
Message msg = new Message(RocketMQConstant.TOPIC_MOMENTS, JSONObject.toJSONString(goods).getBytes(StandardCharsets.UTF_8));
RocketMQUtil.syncSendMsg(defaultMQProducer,msg);
return CommonResponse.success();
}
@GetMapping("/send-async-mq")
public CommonResponse sendMQTest2() throws Exception{
Goods goods = new Goods();
goods.setId(1002);
goods.setCreateTime(new Date());
goods.setUpdateTime(new Date());
goods.setGoodsName("苹果12");
goods.setPrice(new BigDecimal("5999.9"));
goods.setSupply(100);
DefaultMQProducer defaultMQProducer = (DefaultMQProducer)applicationContext.getBean("momentsProducer");
Message msg = new Message(RocketMQConstant.TOPIC_MOMENTS, JSONObject.toJSONString(goods).getBytes(StandardCharsets.UTF_8));
RocketMQUtil.asyncSendMsg(defaultMQProducer,msg);
return CommonResponse.success();
}
}
