前言

最新做了一个消息通知服务，涉及到了消息监听，消息封装，消息发送等方面，今天介绍一下我们引用的消息模板处理，供大家参考。

业务场景

• 即时通知（收到消息后立即通知）
• 延时通知（指定时段通知）

流程图

消息服务监听消息队列消息，判断消息是需要实时通知（立即发送），定时通知（在指定时间发送），实时通知则立即调用发送业务进行消息发送（需要根据业务场景的消息量进行具体设计，可以采用批量消费、多线程消费、按照通知的类型拆分成多个topic进行发送及消费，比如短信一个topic、邮件一个topic、推送一个topic等等方式，下面展示定时通知的代码实现）

定时通知

消息实体

public class MessageInfo implements Serializable {

    /**
     * 通知类型 sms mail push
     */
    private String type;

    /**
     * 通知用户id
     */
    private String userId;

    /**
     * 关键信息
     */
    private Map<String, Object> info;

    /**
     * 通知时间（为空的时候实时推送 非空时候定时推送）
     */
    private Long time;

    // 省略get set...
}
复制代码

监听消息

@Component
public class MessageListener {

    @Autowired
    private TimingMessageCache timingMessageCache;

    @Autowired
    private SenderManager senderManager;

    @Autowired
    private TemplateService templateService;

    @KafkaListener(topics = "message", group = "message-server-group")
    public void messageListener(String message, Acknowledgment ack) {
        try {
            MessageInfo messageInfo = JSONUtil.toBean(message, MessageInfo.class);
            Long time = messageInfo.getTime();
            if (time != null) {
                // 定时通知的消息放入定时消息缓存
                timingMessageCache.add(messageInfo);
            } else {
                // 直接发送 获取模板
                String templateCode = messageInfo.getTemplateCode();
                String template = templateService.getTemplateByCode(templateCode);
                String msg = TemplateUtils.buildMessage(template, messageInfo.getInfo());
                String type = messageInfo.getType();
                senderManager.send(messageInfo.getType(), messageInfo.getUserId(), msg);
            }
            ack.acknowledge();
        } catch (Exception e) {
            // 异常日志
        }

    }
}
复制代码

监听业务系统发送的消息，根据time是否为空来判断是实时通知还是定时通知，定时通知放入定时通知缓存，实时通知直接发送，调用模板工具及发送实现

定时消息缓存

@Component
public class TimingMessageCache {

    @Resource
    private RedisTemplate<String, MessageInfo> redisTemplate;

    private static final String TIMING_MESSAGE_PREFIX = "timing_message:";

    /**
     * 添加
     *
     * @param messageInfo
     */
    public void add(MessageInfo messageInfo) {
        Long time = messageInfo.getTime();
        String minute = FastDateFormat.getInstance("yyyy-MM-dd HH:mm").format(time);
        String key = String.join(":", TIMING_MESSAGE_PREFIX, minute);
        redisTemplate.opsForList().leftPush(key, messageInfo);
    }

    /**
     * 消费指定分钟的所有消息 消息量不大的情况下可以使用此方法 否则可能出现oom
     *
     * @param time
     * @return
     */
    public List<MessageInfo> consumeAll(long time) {
        String minute = FastDateFormat.getInstance("yyyy-MM-dd HH:mm").format(time);
        String key = String.join(":", TIMING_MESSAGE_PREFIX, minute);
        List<MessageInfo> messageInfoList = redisTemplate.opsForList().range(key, 0, -1);
        redisTemplate.delete(key);
        return messageInfoList;
    }

    /**
     * 消费指定数目的消息、适合数量较大的场景下进行分批处理
     *
     * @param time
     * @param limit
     * @return
     */
    public List<MessageInfo> consumePart(long time, int limit) {
        List<MessageInfo> messageInfoList = new ArrayList<>();
        for (int i = 0; i < limit; i++) {
            String minute = FastDateFormat.getInstance("yyyy-MM-dd HH:mm").format(time);
            String key = String.join(":", TIMING_MESSAGE_PREFIX, minute);
            MessageInfo messageInfo = redisTemplate.opsForList().rightPop(key);
            if (messageInfo != null) {
                messageInfoList.add(messageInfo);
            } else {
                // 读到空了 结束循环
                break;
            }
        }
        return messageInfoList;
    }

    /**
     * 获取指定时间消息数目
     *
     * @param time
     * @return
     */
    public long size(long time) {
        String minute = FastDateFormat.getInstance("yyyy-MM-dd HH:mm").format(time);
        String key = String.join(":", TIMING_MESSAGE_PREFIX, minute);
        return redisTemplate.opsForList().size(key);
    }
}
复制代码

定时消息缓存，使用redis list进行存储，同一分钟的数据会存放在同一个list中，key示例 timing_message:2021-10-01 08:01

定时消息处理

@Component
@EnableScheduling
public class TimingMessageHandle {

    @Autowired
    private TimingMessageCache timingMessageCache;

    @Autowired
    private SenderManager senderManager;

    @Autowired
    private TemplateService templateService;

    /**
     * 消费定时消息
     */
    @Scheduled(cron = "0 */1 * * * ?")
    @Async
    public void handle() {
        Calendar calendar = Calendar.getInstance();
        long time = calendar.getTimeInMillis();
        consume(time);
    }

    /**
     * 根据消费对应时间缓存中的消息
     *
     * @param time
     */
    public void consume(long time) {
        long size = timingMessageCache.size(time);
        // 低于1000条 一次性消费完 这个1000可以进行配置 根据服务器性能进行调整
        if (size < 1000) {
            fewConsume(time);
        } else {
            // 数据量大 分批消费 每次消费1000条 解决总共条数非1000整数倍的情况 类似于分页读取最后一页的情况
            for (long i = 0, limit; i < size; i += limit) {
                limit = i + 1000 > size ? size - i : 1000;
                send(timingMessageCache.consumePart(time, limit));
            }
        }
    }


    /**
     * 消息量少的场景调用此方法
     */
    private void fewConsume(long time) {
        List<MessageInfo> messageInfoList = timingMessageCache.consumeAll(time);
        send(messageInfoList);
    }

    /**
     * 发送
     *
     * @param messageInfoList
     */
    private void send(List<MessageInfo> messageInfoList) {
        for (MessageInfo messageInfo : messageInfoList) {
            // 根据模板编码获取对应模板
            String templateCode = messageInfo.getTemplateCode();
            // 根据模板编码获取模板
            String template = templateService.getTemplateByCode(templateCode);
            // 根据模板和消息关键数据生成消息内容
            String msg = TemplateUtils.buildMessage(template, messageInfo.getInfo());
            // 根据类型调用对应发送方法 短信、邮件、推送...
            senderManager.send(messageInfo.getType(), messageInfo.getUserId(), msg);
        }
    }

}
复制代码

前面将定时消息按照前缀+分钟时间作为缓存key，所以定时任务每分钟执行一次，用当前时间生成key去读取缓存，比如指定2021-10-01 08:10发送的消息，定时任务在2021-10-01 08:10就可以从缓存中读取到这些消息并处理掉。

发送实现（伪代码）

发送实现采用接口多实现，指定名称的方式处理，具体可以参考文章

文章链接

// 接口
public interface MessageSender {
    void send(String userId, String msg);
}

// 短信发送实现
@Component("sms")
public class SmsSender implements MessageSender {
    @Override
    public void send(String userId, String msg) {
        // 获取用户手机号
        // 调用三方短信api发送短信
        // 发送短信
    }
}
// 邮件发送实现
@Component("mail")
public class MailSender implements MessageSender {
    @Override
    public void send(String userId, String msg) {
        // 获取用户邮箱
        // 调用邮箱api发送邮件
    }
}
// 管理类 发送统一入口
@Component
public class SenderManager {

    private static final Logger LOGGER = LoggerFactory.getLogger(SenderManager.class);

    /**
     * 空实现避免获取不到对应发送实现出现空指针异常 同时可以打印日志
     */
    private static final MessageSender EMPTY_RECEIVER = (userId, msg) -> {
        LOGGER.info("none sender, userId = {}, msg = {}", userId, msg);
    };

    @Autowired
    private Map<String, MessageSender> senderMessagesMap;

    /**
     * 根据类型调用对应发送实现
     *
     * @param type
     * @param userId
     * @param msg
     */
    public void send(String type, String userId, String msg) {
        senderMessagesMap.get(type).send(userId, msg);
    }
}
复制代码

模板相关（伪代码）

// 模板管理业务 提供根据模板code获取木板的方法
@Service
public class TemplateService {

    public String getTemplateByCode(String templateCode) {
        // 模板一般使用数据库进行存储 如果模板较少且改动得少 使用枚举进行管理也可以
        return "";
    }
}
// 模板工具类
public class TemplateUtils {

    private TemplateUtils() {
    }

    public static String buildMessage(String template, Map<String, Object> info) {
        ST st = new ST(template, '{', '}');
        st.add("info", info);
        return st.render();
    }
}
复制代码

模板的管理可以通过数据库管理或者配置文件，具体实现可以根据自己场景，模板code需要固定下来（约定大于配置），这样在业务系统中根据业务将对应模板code写死即可，在消息通知服务只要不改模板编码，随时调整模板都可以；下面部分介绍模板处理使用的依赖及使用fan

模板处理

通过几个demo演示模板操作的使用方法，供大家参考

示例模板：尊敬的客户，您购买的xxx商品已发货，预计在xxx时间到达目的地，请您注意查收。

引入ST4

<!-- 项目引入ST依赖 -->
<dependency>
    <groupId>org.antlr</groupId>
    <artifactId>ST4</artifactId>
    <version>4.0.8</version>
    <scope>compile</scope>
</dependency>
复制代码

示例代码

public static void main(String[] args) {
    // 模板需要调整成ST支持格式
    String template = "尊敬的客户，您购买的<info.goodsName>商品已发货，预计在<info.arrivalTime>时间到达目的地，请您注意查收。";
    // 数据使用map或实体进行封装都可以
    Map<String, Object> info = new HashMap<>(2);
    info.put("goodsName", "iPhone 13 Pro Max 1T 远峰蓝色");
    info.put("arrivalTime", "2021-10-01");
    ST st = new ST(template, '<', '>');
    st.add("info", info);
    System.out.println(st.render());
}
复制代码

ST对于模板变量占位方式比较灵活，但是这一块需要约定好，变量内容得边界使用'{}'还是'<>'都行，变量内容需要是xxx.xxx，因为我们复杂的模板数据可能来自多条不同数据、比如一个物流信息的数据可能来自于用户基础信息和订单信息及物流信息，这多个数据可能有相同的变量名，具体示例入下：

public static void main(String[] args) {
    String template = "尊敬的{user.name}，您购买的{goods.name}商品已经发货，物流编号为{logistics.number}，预计到货时间为{logistics.arrivalTime}，请您注意按时查收。";
    Map<String, Object> goods = new HashMap<>(2);
    goods.put("name", "iPhone 13 Pro Max 1T 远峰蓝色");
    goods.put("price", 12999);

    Map<String, Object> user = new HashMap<>(2);
    user.put("name", "杨女士");
    user.put("age", 38);

    Map<String, Object> logistics = new HashMap<>(2);
    logistics.put("arrivalTime", "2021-10-01");
    logistics.put("destination", "杭州");
    logistics.put("number", "SF10000001");
    ST st = new ST(template, '{', '}');
    st.add("user", user);
    st.add("goods", goods);
    st.add("logistics", logistics);
    System.out.println(st.render());
}
复制代码

总结

一个消息通知服务技术难度并不大，更多的是结合自己的业务场景，业务量来进行设计，在微服务场景下，一个好用且灵活的通知服务能够减轻整个业务系统在消息通知业务上的开发投入，将定时通知的业务处理交给通知服务也是一个可行的方案，结合redis list + 定时任务，技术方案简单，业务灵活性强，业务系统需要给不同用户在不同时间点发送消息，只需要在一个时间点将所有的消息发送到消息系统，比如有100个用户开启了每日通知，且他们通知时间不同，业务系统只需要启动一个每日执行一次的定时任务，一次将所有的通知发送给消息系统，消息系统自行根据通知时间进行通知，这样对于业务系统来说业务更加简单，所有的通知都可以通过一个每日执行一次的定时任务全部发送到消息系统即可，且任务频率降低了，消息通知的压力只需要在消息服务处理即可。

升级空间

• kafka修改为批量消费，https://juejin.cn/post/6995746569580445709
• topic分区消费，消息服务分布式部署https://juejin.cn/post/6995746569580445709
• 拆分不同类型的通知为多个topic，如message-sms message-mail message-push，分别消费处理
• 发送部分升级为多线程处理