加依赖

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.0.3</version>
</dependency>

写配置

#rocketmq
rocketmq:
  name-server: 127.0.0.1:9876
#  producer:
#    group: test-group #必须制定一个group

编写功能代码

package com.itmuch.usercenter.rocketmq;

import com.itmuch.usercenter.dao.bonusEventLog.BonusEventLogMapper;
import com.itmuch.usercenter.dao.user.UserMapper;
import com.itmuch.usercenter.domain.dto.message.messaging.UserAddBonusMsgDTO;
import com.itmuch.usercenter.domain.entity.bonusEventLog.BonusEventLog;
import com.itmuch.usercenter.domain.entity.user.User;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Date;

@Service //交给spring管理
@RocketMQMessageListener(consumerGroup = "consumer-group", topic = "add-bonus") //生产者的group放在yml中，消费者放在注解里面
public class AddBonusListener implements RocketMQListener<UserAddBonusMsgDTO> {

    @Resource
    private UserMapper userMapper;

    @Resource
    private BonusEventLogMapper bonusEventLogMapper;

    @Override
    public void onMessage(UserAddBonusMsgDTO userAddBonusMsgDTO) {
        //当收到消息的时候执行的业务
        //1、为用户加积分
        Integer id = userAddBonusMsgDTO.getId();
        Integer bonus = userAddBonusMsgDTO.getBonus();
        User user = this.userMapper.selectByPrimaryKey(id);
        user.setBonus(user.getBonus() + bonus);
        this.userMapper.updateByPrimaryKeySelective(user);
        //2、记录日志到bonus_event_log表里面
        this.bonusEventLogMapper.insert(
                BonusEventLog.builder()
                .userId(id)
                .value(bonus)
                .event("CONTRIBUTE")
                .createTime(new Date())
                .description("投稿加积分")
                .build()
        );
    }
}

整体流程：

内容中心通过审核用户投稿进行更新投稿（share）的状态,将用户ID和积分封装到消息体中发送到rocketMQ中的topic（也就是这里的add-bonus）,而用户中心监听rocketmq中的topic（add-bonus）来获取消息体，进行更新用户积分以及日志更新。