统一消息系统

小李：最近我们学校要进行一次大规模的通知发布，比如考试安排、课程调整等，需要向所有学生发送消息。我听说你们部门正在优化消息发送的流程，能不能跟我讲讲？

张工：是的，我们现在正在推进“消息中台”的建设。它可以帮助我们统一管理各种消息渠道，比如短信、邮件、App推送等，而且能支持高并发的批量发送。

小李：听起来很厉害！那这个消息中台具体是怎么工作的呢？有没有什么具体的代码可以参考？

张工：当然有。我们可以用Spring Boot来搭建一个简单的消息中台服务。首先，我们需要定义一个消息模型，然后通过消息队列（比如RabbitMQ或Kafka）来处理批量任务。

小李：那我可以先看看这个消息模型的代码吗？

张工：好的，这是消息模型的一个简单实现：

public class Message {
    private String content;
    private String recipient;
    private String channel; // 消息通道：sms, email, app
    private Date sendTime;

    // 构造函数、getter和setter方法
}
    

小李：明白了。那如何将这些消息批量发送出去呢？是不是需要一个消息队列来缓冲？

张工：没错。我们可以使用RabbitMQ作为消息队列，把消息放入队列后，由多个消费者同时处理。这样可以大大提高发送效率。

小李：那具体怎么实现呢？有没有示例代码？

张工：这里是一个简单的生产者代码，用于将消息发送到队列中：

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

public class MessageProducer {

    private final RabbitTemplate rabbitTemplate;

    public MessageProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public void sendMessage(Message message, String queueName) {
        MessageProperties props = new MessageProperties();
        props.setDeliveryMode(MessageProperties.DELIVERY_MODE_PERSISTENT);
        Message msg = new Message(message.getContent().getBytes(), props);
        rabbitTemplate.send(queueName, msg);
    }
}
    

小李：那消费者那边是怎么处理的呢？

张工：消费者会从队列中取出消息，并根据不同的渠道进行发送。例如，如果是短信，就调用短信接口；如果是邮件，就调用邮件API。下面是一个消费者示例：

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class MessageConsumer {

    @RabbitListener(queues = "message_queue")
    public void receiveMessage(byte[] message) {
        String content = new String(message);
        // 这里可以根据实际内容做进一步解析
        System.out.println("Received message: " + content);
        // 根据channel字段决定发送方式
        // 调用相应的发送接口
    }
}
    

小李：那如果要支持批量发送，应该怎么处理？

张工：我们可以通过分批次的方式进行处理。比如，每次从数据库中读取一定数量的消息，然后打包发送。这样可以避免一次性加载太多数据导致内存溢出。

小李：那有没有具体的代码示例？

张工：当然，下面是一个简单的批量发送逻辑，使用Spring Data JPA来查询消息数据：

import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Service;

@Service
public class MessageService {

    private final MessageRepository messageRepository;

    public MessageService(MessageRepository messageRepository) {
        this.messageRepository = messageRepository;
    }

    public void batchSendMessages(int pageSize, int pageNum) {
        Page messages = messageRepository.findAll(PageRequest.of(pageNum, pageSize));
        for (Message message : messages) {
            // 根据channel发送消息
            if ("sms".equals(message.getChannel())) {
                sendSms(message.getRecipient(), message.getContent());
            } else if ("email".equals(message.getChannel())) {
                sendEmail(message.getRecipient(), message.getContent());
            }
            // 其他渠道处理...
        }
    }

    private void sendSms(String recipient, String content) {
        // 调用短信接口
    }

    private void sendEmail(String recipient, String content) {
        // 调用邮件接口
    }
}
    

小李：这样就能实现批量发送了，对吧？

张工：是的，但为了提高性能，我们还可以引入异步处理。比如，使用线程池或者CompletableFuture来并行处理消息。

小李：那线程池怎么配置呢？

张工：可以使用Spring的@Async注解，结合ThreadPoolTaskExecutor来实现异步处理。下面是一个简单的配置示例：

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

@Configuration
@EnableAsync
public class AsyncConfig {

    @Bean(name = "messageExecutor")
    public ThreadPoolTaskExecutor messageExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(500);
        executor.setThreadNamePrefix("Message-");
        executor.initialize();
        return executor;
    }
}
    

小李：那在业务逻辑中怎么使用这个线程池呢？

张工：可以在服务类中添加@Async注解，然后调用异步方法。例如：

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

@Service
public class MessageService {

    @Async("messageExecutor")
    public void asyncSendMessage(Message message) {
        if ("sms".equals(message.getChannel())) {
            sendSms(message.getRecipient(), message.getContent());
        } else if ("email".equals(message.getChannel())) {
            sendEmail(message.getRecipient(), message.getContent());
        }
        // 其他渠道处理...
    }
}
    

小李：这样就可以实现异步批量发送了，对吧？

张工：是的，这样不仅提高了发送效率，还能有效防止系统因大量请求而崩溃。

小李：那有没有考虑过消息重试和失败处理？

张工：当然有。我们可以使用消息队列的重试机制，或者在发送失败后记录日志，定时重新尝试发送。

小李：那具体的错误处理代码是怎样的？

张工：下面是一个简单的失败处理示例，使用try-catch块捕获异常，并记录日志：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

@Service
public class MessageService {

    private static final Logger logger = LoggerFactory.getLogger(MessageService.class);

    public void sendSingleMessage(Message message) {
        try {
            if ("sms".equals(message.getChannel())) {
                sendSms(message.getRecipient(), message.getContent());
            } else if ("email".equals(message.getChannel())) {
                sendEmail(message.getRecipient(), message.getContent());
            }
            // 记录发送成功
            logger.info("Message sent to {} via {}", message.getRecipient(), message.getChannel());
        } catch (Exception e) {
            logger.error("Failed to send message to {} via {}", message.getRecipient(), message.getChannel(), e);
            // 可以将失败信息存入数据库，后续重试
        }
    }
}
    

小李：看来这个消息中台确实能很好地支持我们学校的批量发消息需求。

张工：是的，而且随着系统的扩展，我们还可以加入更多功能，比如消息模板管理、发送时间调度、发送状态监控等。

小李：那接下来我们是不是需要做一些测试？

张工：是的，我们可以使用JMeter或者Postman来模拟大批量消息发送，测试系统的性能和稳定性。

小李：好的，感谢你的讲解，我现在对消息中台有了更深入的理解。

张工：不客气，如果有其他问题，随时来找我。