Springboot微服務項目整合Kafka實現文章上下架功能
前言：
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內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！
推薦閱讀：
- SpringBoot集成Kafka的步驟
- springboot之配置雙kafka全過程
- Java分佈式學習之Kafka消息隊列
- Springboot集成kafka高級應用實戰分享
- Java實現Kafka生產者和消費者的示例