Springboot微服務專案整合Kafka如何實現文章上下架功能

一：Kafka訊息發送快速入門

1.傳遞字串訊息

（1）發送訊息

建立一個Controller套件並編寫一個測試類別用於發送訊息

package com.my.kafka.controller;
 
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
 
@RestController
public class HelloController {
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;
 
    @GetMapping("hello")
    public String helloProducer(){
        kafkaTemplate.send("my-topic","Hello~");
        return "ok";
    }
}

（2）監聽訊息

編寫測試類別用於接收訊息：

package com.my.kafka.listener;
 
import org.junit.platform.commons.util.StringUtils;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
 
@Component
public class HelloListener {
    @KafkaListener(topics = "my-topic")
    public void helloListener(String message) {
        if(StringUtils.isNotBlank(message)) {
            System.out.println(message);
        }
    }
}

（3）測試結果

開啟瀏覽器輸入localhost:9991/hello，然後到控制台查看訊息，可以看到成功訊息監聽到並且進行了消費。

2.傳遞物件訊息

目前springboot整合後的kafka，因為序列化器是StringSerializer，這個時候如果需要傳遞物件可以有兩種方式：

方式一：可以自訂序列化器，物件類型眾多，這種方式通用性不強，這裡不做介紹。

方式二：可以把要傳遞的物件進行轉json字串，接收訊息後再轉為物件即可，本專案採用這種方式。

（1）修改生產者程式碼

@GetMapping("hello")
public String helloProducer(){
    User user = new User();
    user.setName("赵四");
    user.setAge(20);
    kafkaTemplate.send("my-topic", JSON.toJSONString(user));
    return "ok";
}

（2）結果測試

#可以看到成功接收都物件參數，後期要使用該物件只需要將其轉換成User物件即可。

二：功能引入

1.需求分析

發布文章之後，可能會因為文章出現某些錯誤或其他原因，我們會在文章管理端實作文章的上下架功能（見下圖），也即當管理端實現對文章下架之後移動端將不會再展示該文章，只有該文章重新被上架之後才能在移動端看到該文章資訊。

2.邏輯分析

#後端接收到前端傳過來的參數之後要先做一個校驗，參數不為空才能繼續往下執行，首先應該根據前端傳過來的文章id（自媒體端文章id）查詢自媒體資料庫的文章資訊並判斷該文章是否已是發布狀態，因為只有審核成功並成功發布了的文章才能進行上下架操作。自媒體端微服務對文章上下架狀態進行修改之後便可以向Kafka發送一條消息，該消息為Map對象，裡面存儲的數據為移動端的文章id以及前端傳過來的上下架參數enable，當然要將該Map物件轉換成JSON字串才能進行傳送。

文章微服務監聽到Kafka發送過來的訊息之後將JSON字串轉換成Map物件之後再取得相關參數對行動端文章的上下架狀態進行修改。

三：前期準備

1.引入依賴

<!-- kafkfa -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
</dependency>

2.定義常數

package com.my.common.constans;
public class WmNewsMessageConstants {
    public static final String WM_NEWS_UP_OR_DOWN_TOPIC="wm.news.up.or.down.topic";
}

3.Kafka配置資訊

由於我是用Nacos來作為註冊中心，所以配置資訊放置在Nacos上面即可。

（1）自媒體端設定

spring:
  kafka:
    bootstrap-servers: 4.234.52.122:9092
    producer:
      retries: 10
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

（2）行動端設定

spring:
  kafka:
    bootstrap-servers: 4.234.52.122:9092
    consumer:
      group-id: ${spring.application.name}-test
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

四：程式碼實作

1.自媒體端

@Autowired
private KafkaTemplate<String,String> kafkaTemplate;
/**
 * 文章下架或上架
 * @param id
 * @param enable
 * @return
 */
@Override
public ResponseResult downOrUp(Integer id,Integer enable) {
    log.info("执行文章上下架操作...");
    if(id == null || enable == null) {
        return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
    }
    //根据id获取文章
    WmNews news = getById(id);
    if(news == null) {
        return ResponseResult.errorResult(AppHttpCodeEnum.DATA_NOT_EXIST,"文章信息不存在");
    }
    //获取当前文章状态
    Short status = news.getStatus();
    if(!status.equals(WmNews.Status.PUBLISHED.getCode())) {
        return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID,"文章非发布状态，不能上下架");
    }
 
    //更改文章状态
    news.setEnable(enable.shortValue());
    updateById(news);
    log.info("更改文章上架状态{}-->{}",status,news.getEnable());
 
    //发送消息到Kafka
    Map<String, Object> map = new HashMap<>();
    map.put("articleId",news.getArticleId());
    map.put("enable",enable.shortValue());
    kafkaTemplate.send(WmNewsMessageConstants.WM_NEWS_UP_OR_DOWN_TOPIC,JSON.toJSONString(map));
    log.info("发送消息到Kafka...");
 
    return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);
}

2.行動裝置

（1）設定監聽器

package com.my.article.listener;
 
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.my.article.service.ApArticleService;
import com.my.common.constans.WmNewsMessageConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.kafka.annotation.KafkaListener;
 
 
@Slf4j
@Component
public class EnableListener {
    @Autowired
    private ApArticleService apArticleService;
 
    @KafkaListener(topics = WmNewsMessageConstants.WM_NEWS_UP_OR_DOWN_TOPIC)
    public void downOrUp(String message) {
        if(StringUtils.isNotBlank(message)) {
            log.info("监听到消息{}",message);
            apArticleService.downOrUp(message);
        }
    }
}

（2）取得訊息並修改文章狀態

/**
* 文章上下架
* @param message
* @return
*/
@Override
public ResponseResult downOrUp(String message) {
    Map map = JSON.parseObject(message, Map.class);
    //获取文章id
    Long articleId = (Long) map.get("articleId");
    //获取文章待修改状态
    Integer enable = (Integer) map.get("enable");
    //查询文章配置
    ApArticleConfig apArticleConfig = apArticleConfigMapper.selectOne
            (Wrappers.<ApArticleConfig>lambdaQuery().eq(ApArticleConfig::getArticleId, articleId));
    if(apArticleConfig != null) {
        //上架
        if(enable == 1) {
            log.info("文章重新上架");
            apArticleConfig.setIsDown(false);
            apArticleConfigMapper.updateById(apArticleConfig);
        }
        //下架
        if(enable == 0) {
            log.info("文章下架");
            apArticleConfig.setIsDown(true);
            apArticleConfigMapper.updateById(apArticleConfig);
        }
    }
    else {
        throw new RuntimeException("文章信息不存在");
    }
    return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);
}

以上是Springboot微服務專案整合Kafka如何實現文章上下架功能的詳細內容。更多資訊請關注PHP中文網其他相關文章！