从 0 到 1 搞懂：Feign/RocketMQ/自定义线程池全指南

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 自定义线程池配置 = 给小卖部雇临时工，定好规矩
 */
@Configuration
public class ThreadPoolConfig {

    @Bean("myTaskExecutor")
    public Executor myTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 核心线程数 = 长期雇的固定临时工，就算没事干也不辞退
        executor.setCorePoolSize(5);
        // 最大线程数 = 忙的时候最多能雇多少临时工
        executor.setMaxPoolSize(10);
        // 队列容量 = 临时工都忙的时候，活放在等候区的最大数量
        executor.setQueueCapacity(100);
        // 非核心线程空闲时间 = 临时雇的临时工，没事干多久就辞退
        executor.setKeepAliveSeconds(60);
        // 线程名前缀 = 给临时工起名字，方便找是谁干的活
        executor.setThreadNamePrefix("my-task-thread-");
        // 拒绝策略 = 临时工全忙、等候区也满了，新活怎么处理（这里是交给主线程干）
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 初始化
        executor.initialize();
        return executor;
    }
}

import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;

@Service
public class GoodsService {

    // 注入我们配置好的线程池
    @Resource(name = "myTaskExecutor")
    private Executor taskExecutor;

    // 模拟商品详情页查询：并行查多个数据源，提速返回
    public GoodsDetailVO getGoodsDetail(Long goodsId) {
        // 1. 并行查商品基础信息，交给临时工干
        Future<GoodsBaseInfo> baseInfoFuture = taskExecutor.submit(() -> getGoodsBaseInfo(goodsId));
        // 2. 并行查库存，交给临时工干
        Future<GoodsStock> stockFuture = taskExecutor.submit(() -> getGoodsStock(goodsId));
        // 3. 并行查价格，交给临时工干
        Future<GoodsPrice> priceFuture = taskExecutor.submit(() -> getGoodsPrice(goodsId));

        try {
            // 4. 汇总所有临时工的干活结果
            GoodsDetailVO vo = new GoodsDetailVO();
            vo.setBaseInfo(baseInfoFuture.get()); // 拿基础信息结果
            vo.setStock(stockFuture.get()); // 拿库存结果
            vo.setPrice(priceFuture.get()); // 拿价格结果
            return vo;
        } catch (Exception e) {
            // 异常处理
            throw new RuntimeException("查询商品详情失败", e);
        }
    }

    // 以下是模拟的三个查询方法，实际项目中是查数据库/调用其他接口
    private GoodsBaseInfo getGoodsBaseInfo(Long goodsId) throws InterruptedException {
        // 模拟查库耗时100ms
        Thread.sleep(100);
        return new GoodsBaseInfo();
    }

    private GoodsStock getGoodsStock(Long goodsId) throws InterruptedException {
        Thread.sleep(100);
        return new GoodsStock();
    }

    private GoodsPrice getGoodsPrice(Long goodsId) throws InterruptedException {
        Thread.sleep(100);
        return new GoodsPrice();
    }
}

<!-- SpringBoot整合RocketMQ的依赖 -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.3</version>
</dependency>

rocketmq:
  # RocketMQ的地址，就是快递驿站的地址
  name-server: 127.0.0.1:9876
  # 生产者配置
  producer:
    # 生产者组名，就是卖家的店铺编号
    group: note-producer-group

import lombok.Data;
import java.io.Serializable;

/**
 * 笔记缓存消息体 = 快递包裹里的东西
 */
@Data
public class NoteCacheMsg implements Serializable {
    private static final long serialVersionUID = 1L;

    // 笔记ID
    private Long noteId;
    // 笔记详情数据，要写入Redis的内容
    private NoteDetailRspVO noteDetail;
}

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;

@Service
public class NoteService {

    // 注入RocketMQ模板，就是寄快递的工具
    @Resource
    private RocketMQTemplate rocketMQTemplate;

    // 查询笔记详情的核心接口
    public Response<NoteDetailRspVO> getNoteDetail(Long noteId) {
        // 1. 核心业务：查数据库，获取笔记详情
        NoteDetailRspVO noteDetail = queryNoteDetailFromDB(noteId);

        // 2. 构建消息包裹
        NoteCacheMsg msg = new NoteCacheMsg();
        msg.setNoteId(noteId);
        msg.setNoteDetail(noteDetail);

        // 3. 把包裹交给快递驿站（发消息到RocketMQ），发完就走，不阻塞
        // note-cache-topic 就是收件地址，只有对应地址的消费者能收到
        rocketMQTemplate.syncSend("note-cache-topic", msg);

        // 4. 直接返回结果给用户，不用等缓存写入完成
        return Response.success(noteDetail);
    }

    // 模拟从数据库查笔记详情
    private NoteDetailRspVO queryNoteDetailFromDB(Long noteId) {
        // 实际项目中是查数据库
        return new NoteDetailRspVO();
    }
}

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import cn.hutool.core.util.RandomUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * 笔记缓存消费者 = 收件人，专门收note-cache-topic地址的包裹
 */
@Component
// topic：收件地址，consumerGroup：收件人编号
@RocketMQMessageListener(
        topic = "note-cache-topic",
        consumerGroup = "note-cache-consumer-group"
)
public class NoteCacheConsumer implements RocketMQListener<NoteCacheMsg> {

    @Resource
    private RedisTemplate<String, String> redisTemplate;

    @Resource
    private ObjectMapper objectMapper;

    // 收到消息（包裹）后，执行的逻辑
    @Override
    public void onMessage(NoteCacheMsg msg) {
        try {
            // 1. 把笔记详情转成JSON字符串
            String noteDetailJson = objectMapper.writeValueAsString(msg.getNoteDetail());
            // 2. 生成Redis的key
            String redisKey = "note:detail:" + msg.getNoteId();
            // 3. 过期时间：保底1天 + 随机秒数，打散过期时间，防止缓存雪崩
            long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);
            // 4. 写入Redis
            redisTemplate.opsForValue().set(redisKey, noteDetailJson, expireSeconds, TimeUnit.SECONDS);
        } catch (Exception e) {
            // 异常处理：抛出异常，RocketMQ会自动重试，不会丢消息
            throw new RuntimeException("缓存写入失败", e);
        }
    }
}

@Service
public class NoteService {

    @Resource
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Resource
    private RedisTemplate<String, String> redisTemplate;

    public Response<NoteDetailRspVO> getNoteDetail(Long noteId) {
        // 1. 查数据库，获取笔记详情
        NoteDetailRspVO findNoteDetailRspVO = queryNoteDetailFromDB(noteId);

        // 2. 异步线程中将笔记详情存入 Redis（反面写法，有坑）
        threadPoolTaskExecutor.submit(() -> {
            String noteDetailJson1 = JsonUtils.toJsonString(findNoteDetailRspVO);
            long expireSeconds = 60*60*24 + RandomUtil.randomInt(60*60*24);
            redisTemplate.opsForValue().set(noteDetailRedisKey, noteDetailJson1, expireSeconds, TimeUnit.SECONDS);
        });

        // 3. 返回结果
        return Response.success(findNoteDetailRspVO);
    }
}