[大数据]Kafka，异步消息系统

阻塞队列

• BlockingQueue（java接口）
  解决线程通信的问题
  阻塞方法：put、take;阻塞时不会占用系统资源
• 生产者消费者模式
• 实现类：
  ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue、SynchronousQueue、DelayQueue;

Kafka入门

kafka是分布式的流媒体平台，应用：消息系统，日志收集，用户行为追踪，流式处理。
kafka特点：高吞吐量，消息持久化，高可靠性（分布式保证），高扩展性。
（对硬盘的读取，顺序读取甚至比对内存的随机读取要快，kafka顺序读取硬盘，保证高吞吐量，高速率）
消息队列两种方式：点对点、发布订阅模式（kafka采用）
Kafka术语：

• Broker（服务器），Zookeeper(用于管理集群)
• Topic(理解为一个文件夹，生产者把消息发布到的位置)、Partition(对topic的分区)、offset(消息在分区内存放的索引)
• Leader Replica(主副本)、Follower Replica（从副本，只是用来备份）

cd到kafka相应的目录下
启动zookeeper服务器：bin\windows\zookeeper-server-start.bat config\zookeeper.properties
启动kafka服务器： bin\windows\kafka-server-start.bat config\server.properties

kafka出错，把kafka-log文件夹删除重新启动试试

Spring整合Kafka

• 引入依赖 ：spring-kafka
• 配置kafka：配置server、consumer

# KafkaProperties
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=community-consumer-group
# 是否自动提交消费者的偏移量
spring.kafka.consumer.enable-auto-commit=true 
# 自动提交的频率
spring.kafka.consumer.auto-commit-interval=3000

• 访问kafka

    //生产者，生产者发消息是主动调用方法去发送的
    kafkaTemplate.send(topic，data) 
    
    //消费者，监听test的topic，阻塞的读
    @kafkaListener(topics = {"test"})  
    // 会把读取的消息封装成ConsumerRecord，从record读原始消息
    public void handlMessage(ConsumerRecord record){}

服务器启动时，Spring Kafka会自动创建主题。但是测试代码中不会，所以测试代码中你需要用的主题，需要手动创建。

@RunWith(SpringRunner.class)
@SpringBootTest
@ContextConfiguration(classes = CommunityApplication.class)
public class KafkaTest {
    @Autowired
    private kafkaProducer kafkaProducer;

    @Test
    public void testkafka(){
        kafkaProducer.senMessage("test","你好");
        kafkaProducer.senMessage("test","在哪");

        try{
            Thread.sleep(1000*10);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
@Component
class kafkaProducer{
    @Autowired
    private KafkaTemplate kafkaTemplate;
    // 发送的主题和内容
    public void senMessage(String topic,String content){
        kafkaTemplate.send(topic,content);
    }
}
@Component
class kafkaConsumer{

    @KafkaListener(topics = {"test"})
    public void handler(ConsumerRecord record){
        System.out.println(record.value());
    }
}

发送系统通知

评论、点赞、关注等产生的事件发生后，消费者线程异步的从消息队列中取出处理，提高并发性

• 触发事件
  评论后，发布通知
  点赞后，发布通知
  关注后，发布通知

//在添加评论的方法中加上如下代码，评论后调用发布信息

//触发评论事件
        Event event = new Event()
                .setTopic(TOPIC_COMMENT)
                .setUserId(hostHolder.getUser().getId())
                .setEntityType(comment.getEntityType())
                .setEntityUserId(comment.getEntityId())
                .setData("postId",discussPostId);
        // 评论的实体的作者EntityUser，帖子的作者，评论的作者，在不同的表中，分开来查
        if(comment.getEntityType() == ENTITY_TYPE_POST){ //帖子的评论
            DiscussPost target = discussPostService.findDiscussPostById(comment.getEntityId());
            event.setEntityUserId(target.getUserId());
        }else if(comment.getEntityType() == ENTITY_TYPE_COMMENT){
            Comment target = commentService.findCommentById(comment.getEntityId());
            event.setEntityUserId(target.getUserId());
        }

        eventProducer.fireEvent(event);

• 处理事件
  封装事件对象
  开发事件的生产者
  开发事件的消费者

消费者监听事件，然后把事件存放进入数据库中

 //消费评论、点赞、关注事件
    @KafkaListener(topics = {TOPIC_COMMENT, TOPIC_LIKE, TOPIC_FOLLOW})
    public void handleCommentMessage(ConsumerRecord record) {
        if (record == null || record.value() == null) {
            logger.error("消息的内容为空!");
            return;
        }

        Event event = JSONObject.parseObject(record.value().toString(), Event.class);
        if (event == null) {
            logger.error("消息格式错误!");
            return;
        }

        // 发送站内通知
        Message message = new Message();
        message.setFromId(SYSTEM_USER_ID);
        message.setToId(event.getEntityUserId());
        message.setConversationId(event.getTopic());
        message.setCreateTime(new Date());

        Map<String, Object> content = new HashMap<>();
        content.put("userId", event.getUserId());
        content.put("entityType", event.getEntityType());
        content.put("entityId", event.getEntityId());

        //遍历map对象，再把key value放到content中
        if (!event.getData().isEmpty()) {
            for (Map.Entry<String, Object> entry : event.getData().entrySet()) {
                content.put(entry.getKey(), entry.getValue());
            }
        }

        //把content转为JSON字符串
        message.setContent(JSONObject.toJSONString(content));
        messageService.addMessage(message);
    }

• 通知列表
  显示评论、点赞、关注三种类型的通知
• 通知详情
  分页显示某一类主题所包含的通知
• 未读消息
  在页面头部显示所有的未读消息数量

 	// 查询某个主题下最新的通知
    Message selectLatestNotice(@Param("userId")int userId, @Param("topic")String topic);

    // 查询某个主题所包含的通知数量
    int selectNoticeCount(@Param("userId")int userId, @Param("topic")String topic);

    // 查询未读的通知的数量
    int selectNoticeUnreadCount(@Param("userId")int userId,@Param("topic") String topic);
 
    <select id="selectLatestNotice" resultType="Message">
        select <include refid="selectFields"></include>
        from message
        where id in (
	        select max(id) from message
	        where status != 2
	        and from_id = 1
	        and to_id = #{userId}
	        and conversation_id = #{topic}
        )
    </select>

    <select id="selectNoticeCount" resultType="int">
        select count(id) from message
        where status != 2
        and from_id = 1
        and to_id = #{userId}
        and conversation_id = #{topic}
    </select>

    <select id="selectNoticeUnreadCount" resultType="int">
        select count(id) from message
        where status = 0
        and from_id = 1
        and to_id = #{userId}
        <if test="topic!=null">
            and conversation_id = #{topic}
        </if>
    </select>