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

2022-07-30 18:05:28

前言：

1.前面基於Springboot的單體專案介紹已經完結了，至於專案中的其他功能實現我這裡就不打算介紹了，因為涉及的知識點不難，而且都是簡單的CRUD操作，假如有興趣的話可以私信我我再看看要不要寫幾篇文章做個介紹。

2.完成上一階段的學習，我就投入到了微服務的學習當中，所用教學為B站上面黑馬的微服務教學。由於我的記性不是很好，所以對於新事物的學習我比較喜歡做筆記以加強理解，在這裡我會將筆記的重點內容做個總結髮布到“微服務學習”筆記欄目中。我是趙四，一名有追求的程式設計師，希望大家能多多支援，能給我點個關注就更好了。

一：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實現文章上下架功能的文章就介紹到這了,更多相關Springboot整合Kafka內容請搜尋it145.com以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援it145.com！