组件配置：recketmq.md · saderqw/points-service-cloud

# Mybatis-plus

[官方文档](https://github.com/apache/rocketmq/tree/master/docs/cn)

[管理端](https://github.com/apache/rocketmq-externals/releases/tag/rocketmq-console-1.0.0)

## 使用配置

### 引入
通过以下maven配置，引入到工程中。
```xml

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <exclusions>
        <exclusion>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </exclusion>
    </exclusions>
    <version>${rocketmq-spring-boot-starter.version}</version>
</dependency>

```

### 配置

```yaml
rocketmq:
    name-server: cloud.rocketmq.server:9876 # 访问地址
    producer:
        group: dev1 # 必须指定group
        send-message-timeout: 3000 # 消息发送超时时长，默认3s
        retry-times-when-send-failed: 3 # 同步发送消息失败重试次数，默认2
        retry-times-when-send-async-failed: 3 # 异步发送消息失败重试次数，默认2
        access-key: rocketmq2
        secret-key: 12345678
```
### 项目引入

```xml
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
</dependency>
```

### 生产者服务示例

```java
package com.djbx.points.member.mq;

import com.djbx.points.member.api.dto.MemberInDTO;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.JsonPrimitive;
import com.google.gson.JsonSerializer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

@Slf4j
@Component
public class MQProducerService {

//	@Value("${rocketmq.producer.send-message-timeout}")
//    private Integer messageTimeOut;

/**
     * 建议正常规模项目统一用一个TOPIC
     */
    private static final String TOPIC = "TOPIC_DEV1";

/**
     * 直接注入使用，用于发送消息到broker服务器
     */
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

/**
     * 普通发送（这里的参数对象inDTO可以随意定义，可以发送个对象，也可以是字符串等）
     */
    public void send(MemberInDTO inDTO) {
        //":member"用于标识TOPIC下的不同分类标记
        Gson gson = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss")
                .registerTypeAdapter(LocalDateTime.class, (JsonSerializer<LocalDateTime>) (src, typeOfSrc, context) -> new JsonPrimitive(src.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))))
                .registerTypeAdapter(LocalDate.class, (JsonSerializer<LocalDate>) (src, typeOfSrc, context) -> new JsonPrimitive(src.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"))))
                .create();
        rocketMQTemplate.convertAndSend(TOPIC + ":member", gson.toJson(inDTO));
    }

//    /**
//     * 发送同步消息（阻塞当前线程，等待broker响应发送结果，这样不太容易丢失消息）
//     * （msgBody也可以是对象，sendResult为返回的发送结果）
//     */
//    public SendResult sendMsg(String msgBody) {
//        SendResult sendResult = rocketMQTemplate.syncSend(TOPIC, MessageBuilder.withPayload(msgBody).build());
//        log.info("【sendMsg】sendResult={}", JSON.toJSONString(sendResult));
//        return sendResult;
//    }
//
//	/**
//     * 发送异步消息（通过线程池执行发送到broker的消息任务，执行完后回调：在SendCallback中可处理相关成功失败时的逻辑）
//     * （适合对响应时间敏感的业务场景）
//     */
//    public void sendAsyncMsg(String msgBody) {
//        rocketMQTemplate.asyncSend(TOPIC, MessageBuilder.withPayload(msgBody).build(), new SendCallback() {
//            @Override
//            public void onSuccess(SendResult sendResult) {
//                // 处理消息发送成功逻辑
//            }
//            @Override
//            public void onException(Throwable throwable) {
//                // 处理消息发送异常逻辑
//            }
//        });
//    }
//
//	/**
//     * 发送延时消息（上面的发送同步消息，delayLevel的值就为0，因为不延时）
//     * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
//     */
//    public void sendDelayMsg(String msgBody, int delayLevel) {
//        rocketMQTemplate.syncSend(TOPIC, MessageBuilder.withPayload(msgBody).build(), messageTimeOut, delayLevel);
//    }
//
//    /**
//     * 发送单向消息（只负责发送消息，不等待应答，不关心发送结果，如日志）
//     */
//    public void sendOneWayMsg(String msgBody) {
//        rocketMQTemplate.sendOneWay(TOPIC, MessageBuilder.withPayload(msgBody).build());
//    }
//
//	/**
//     * 发送带tag的消息，直接在topic后面加上":tag"
//     */
//    public SendResult sendTagMsg(String msgBody) {
//        return rocketMQTemplate.syncSend(TOPIC + ":tag", MessageBuilder.withPayload(msgBody).build());
//    }

}

```

### 消费者服务示例

```java
package com.djbx.points.member.mq;

import com.djbx.points.member.api.dto.MemberInDTO;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.JsonDeserializer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * topic需要和生产者的topic一致，consumerGroup属性是必须指定的，内容可以随意
 * selectorExpression的意思指的就是tag，默认为“*”，不设置的话会监听所有消息
 */
@Slf4j
@Service
@RocketMQMessageListener(topic = "TOPIC_DEV1", selectorExpression = "member", consumerGroup = "DEV1")
public class MQConsumerService implements RocketMQListener<String> {

/**
     * 监听到消息就会执行此方法
     * @param msg
     */
    @Override
    public void onMessage(String msg) {
        Gson gson = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss")
                .registerTypeAdapter(LocalDateTime.class, (JsonDeserializer<LocalDateTime>) (json, type, jsonDeserializationContext) -> {
                    String datetime = json.getAsJsonPrimitive().getAsString();
                    return LocalDateTime.parse(datetime, DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
                }).registerTypeAdapter(LocalDate.class, (JsonDeserializer<LocalDate>) (json, type, jsonDeserializationContext) -> {
                    String datetime = json.getAsJsonPrimitive().getAsString();
                    return LocalDate.parse(datetime, DateTimeFormatter.ofPattern("yyyy-MM-dd"));}).create();
        MemberInDTO memberInDTO = gson.fromJson(msg, MemberInDTO.class);
        log.info("监听到消息：user={}", memberInDTO);
    }

//    /**
    //     * MessageExt：是一个消息接收通配符，不管发送的是String还是对象，都可接收，当然也可以像上面明确指定类型（还是指定类型较方便）
    //     */
    // @Override
//        public void onMessage(MessageExt messageExt) {
//            byte[] body = messageExt.getBody();
//            String msg = new String(body);
//            log.info("监听到消息：msg={}", msg);
//        }

}

```

saderqw/points-service-cloud

简介

发行版

贡献者

近期动态

saderqw/points-service-cloud .gitee-modal { width: 500px !important; }

简介

发行版

贡献者

近期动态

搜索帮助

saderqw/points-service-cloud