Bagaimana projek perkhidmatan mikro Springboot menyepadukan Kafka untuk melaksanakan fungsi muat naik dan nyahsenarai artikel

1: Mula Pantas untuk Menghantar Mesej Kafka

1 Lulus Mesej Rentetan

(1) Hantar Mesej

Buat pakej Pengawal dan tulis kelas ujian untuk Hantar. mesej

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) Dengar mesej

Tulis kelas ujian untuk menerima mesej:

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) Keputusan ujian

Buka input penyemak imbas localhost :9991/hello, dan kemudian pergi ke konsol untuk melihat mesej Anda boleh melihat bahawa mesej yang berjaya telah dipantau dan digunakan.

2. Menghantar mesej objek

Kafka bersepadu springboot, kerana penyeri adalah StringSerializer, terdapat dua cara untuk menghantar objek pada masa ini:

Kaedah 1: Anda boleh menyesuaikan serializer dan terdapat banyak jenis objek Kaedah ini tidak begitu serba boleh dan tidak akan diperkenalkan di sini.

Kaedah 2: Anda boleh menukar objek untuk dipindahkan ke dalam rentetan json, dan kemudian menukarnya menjadi objek selepas menerima mesej Kaedah ini digunakan dalam projek ini.

(1) Ubah suai kod pengeluar

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

(2) Ujian keputusan

Anda boleh melihat bahawa semua parameter objek berjaya diterima , kemudian Untuk menggunakan objek ini, anda hanya perlu menukarnya menjadi objek Pengguna.

Dua: Pengenalan fungsi

1. Analisis keperluan

Selepas menerbitkan artikel, mungkin terdapat beberapa ralat atau sebab lain dalam artikel tersebut sisi pengurusan artikel Fungsi muat naik dan nyahsenarai (lihat gambar di bawah), iaitu, apabila terminal pengurusan mengalih keluar artikel dari rak, terminal mudah alih tidak akan lagi memaparkan artikel Hanya selepas artikel itu disenaraikan semula, artikel itu boleh maklumat boleh dilihat pada terminal mudah alih.

2. Analisis logik

Selepas bahagian belakang menerima parameter yang diluluskan oleh bahagian hadapan, ia mesti terlebih dahulu melakukan pengesahan , parameter tidak kosong sebelum pelaksanaan boleh diteruskan Pertama, maklumat artikel pangkalan data media kendiri harus ditanya berdasarkan id artikel yang diluluskan dari bahagian hadapan (id artikel akhir media kendiri) dan. menilai sama ada artikel itu telah diterbitkan, kerana hanya semakan yang berjaya dan berjaya Hanya artikel yang diterbitkan boleh dimuat naik atau dialih keluar. Selepas perkhidmatan mikro sisi media kendiri mengubah status muat naik dan nyahsenarai artikel, ia boleh menghantar mesej kepada Kafka Mesej ialah objek Peta Data yang disimpan di dalamnya ialah id artikel terminal mudah alih dan parameter daya yang dihantar dari bahagian hadapan Sudah tentu, mesej ini mestilah objek Peta boleh ditukar menjadi rentetan JSON sebelum ia boleh dihantar.

Selepas mendengar mesej yang dihantar oleh Kafka, perkhidmatan mikro artikel menukar rentetan JSON menjadi objek Peta dan kemudian mendapatkan parameter yang berkaitan untuk mengubah suai status atas dan bawah artikel mudah alih.

Tiga: Persediaan awal

1. Memperkenalkan kebergantungan

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

2 Takrifkan pemalar

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.Maklumat konfigurasi Kafka

Due. kepada saya menggunakan Nacos sebagai pusat pendaftaran, jadi maklumat konfigurasi boleh diletakkan pada Nacos.

(1) Konfigurasi terminal media kendiri

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) Konfigurasi terminal mudah alih

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

Empat: Pelaksanaan kod

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. Terminal mudah alih

(1) Sediakan pendengar

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) Dapatkan mesej dan ubah suai status artikel

rreee

Atas ialah kandungan terperinci Bagaimana projek perkhidmatan mikro Springboot menyepadukan Kafka untuk melaksanakan fungsi muat naik dan nyahsenarai artikel. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!