SpringCloud微服务项目实战 - 7.kafka及文章上下架

news2024/9/20 3:06:33

一步一步地苦熬苦掖,终于我们也看见了花团锦簇,我们也知道了彩灯佳话。那一夜,我也曾梦见百万雄兵。

在这里插入图片描述



系列文章目录

  1. 项目搭建
  2. App登录及网关
  3. App文章
  4. 自媒体平台(博主后台)
  5. 内容审核(自动)
  6. 延迟任务

文章目录

  • 系列文章目录
  • 一、Kafka
    • 1. 概述
      • ⑴. 消息中间件对比
      • ⑵. 介绍
    • 2. 环境搭建
    • 3. 入门案例
      • ⑴. 模块搭建
      • ⑵. 生产者发送消息
      • ⑶. 消费者接收消息
      • ⑷. 测试
      • ⑸. 消费者组
        • ①. 一对一
        • ②. 一对多
    • 4. 分区机制
    • 5. 高可用设计
      • ⑴. 集群
      • ⑵. 备份机制(Replication)
    • 6. 生产者详解
      • ⑴. 发送类型
      • ⑵. 参数详解
        • ①. 消息确认机制
        • ②. 重试次数
        • ③. 消息压缩
    • 7. 消费者详解
      • ⑴. 消费者组
      • ⑵. 消息有序性
      • ⑶. 手动提交偏移量
        • ①. 正常的情况
        • ②. 问题一
        • ③. 问题二:
        • ④. 自动提交偏移量
  • 二、自媒体文章上下架
    • 1. Spring集成kafka
      • ⑴. Demo搭建
        • ①. pom依赖
        • ②. 启动类
        • ③. 配置信息
        • ④. 生产者
        • ⑤. 消费者
        • ⑥. 测试
      • ⑵. 传递消息为对象
        • ①. user对象
        • ②. 生产者
        • ③. 消费者
        • ④. 测试
    • 2. 需求分析
    • 3. 自媒体端(功能实现)
      • ⑴. 接口定义
        • ①. 说明
        • ②. DTO
        • ③. ResponseResult
      • ⑵. 接口实现
        • ①. DTO
        • ②. Controller
        • ③. Service
        • ④. ServiceImpl
        • ⑤. 测试
    • 4. Kafka(传递消息)
      • ⑴. 公共
        • ①. 常量
        • ②. pom依赖
      • ⑵. 发送消息
        • ①. Nacos
        • ②. 文章上下架后发送消息
      • ⑶. 接收消息
        • ①. Nacos
        • ②. 文章上下架后接收消息
    • 5. 文章端(修改文章信息)
        • ①. Service
        • ②. ServiceImpl
        • ③. 修改文章配置
        • ④. 测试




一、Kafka

1. 概述

⑴. 消息中间件对比

特性ActiveMQRabbitMQRocketMQKafka
开发语言javaerlangjavascala
单机吞吐量万级万级10万级100万级
时效性msusmsms级以内
可用性高(主从)高(主从)非常高(分布式)非常高(分布式)
功能特性成熟的产品、较全的文档、各种协议支持好并发能力强、性能好、延迟低MQ功能比较完善,扩展性佳只支持主要的MQ功能,主要应用于大数据领域

选择建议:

消息中间件建议
Kafka追求高吞吐量,适合产生大量数据的互联网服务的数据收集业务
RocketMQ可靠性要求很高的金融互联网领域,稳定性高,经历了多次阿里双11考验
RabbitMQ性能较好,社区活跃度高,数据量没有那么大,优先选择功能比较完备的RabbitMQ

⑵. 介绍

在这里插入图片描述
Kafka 是一个分布式流媒体平台,类似于消息队列或企业消息传递系统。kafka官网:http://kafka.apache.org/

名词解释
在这里插入图片描述

  • producer: 发布消息的对象称之为主题生产者(Kafka topic producer)
  • topic: Kafka将消息分门别类,每一类的消息称之为一个主题(Topic)
  • consumer: 订阅消息并处理发布的消息的对象称之为主题消费者(consumers)
  • broker: 已发布的消息保存在一组服务器中,称之为Kafka集群。集群中的每一个服务器都是一个代理(Broker)。 消费者可以订阅一个或多个主题(topic),并从Broker拉数据,从而消费这些已发布的消息。

2. 环境搭建

Kafka对于zookeeper是强依赖,保存kafka相关的节点数据,所以安装Kafka之前必须先安装zookeeper

zookeeper:

# 下载镜像(已经完成)
docker pull zookeeper:3.4.14
# 查看本地镜像
docker images
# 创建容器
docker run -d --name zookeeper -p 2181:2181 zookeeper:3.4.14

kafka:

# 下载镜像
docker pull wurstmeister/kafka:2.12-2.3.1
# 创建容器
docker run -d --name kafka \
--env KAFKA_ADVERTISED_HOST_NAME=192.168.200.130 \
--env KAFKA_ZOOKEEPER_CONNECT=192.168.200.130:2181 \
--env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.200.130:9092 \
--env KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
--env KAFKA_HEAP_OPTS="-Xmx256M -Xms256M" \
--net=host wurstmeister/kafka:2.12-2.3.1
# 列出容器
docker ps
# 查看容器日志
docker logs -f 【容器ID】

3. 入门案例

⑴. 模块搭建

在这里插入图片描述
配置 kafka 依赖:
编辑 heima-leadnews-test/kafka-demo/pom.xml 文件:

    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
        </dependency>
    </dependencies>

⑵. 生产者发送消息

新建 heima-leadnews-test/kafka-demo/src/main/java/com/heima/kafka/sample/ProducerQuickStart.java 文件:

/**
 * 生产者
 */
public class ProducerQuickStart {
    public static void main(String[] args) {

        // 1. kafka链接配置信息
        Properties prop = new Properties();
        // 1.1 kafka链接地址
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");
        // 1.2 key 和 value 的序列化
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // 2. 创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(prop);

        // 3. 发送消息
        ProducerRecord<String, String> kvProducerRecord = new ProducerRecord<String, String>("topic-first", "key-001", "hello kafka");
        producer.send(kvProducerRecord);

        // 4. 关闭消息通道(必须关闭,否则信息发送不成功)
        producer.close();
    }
}

⑶. 消费者接收消息

新建 heima-leadnews-test/kafka-demo/src/main/java/com/heima/kafka/sample/ConsumerQuickStart.java 文件:

/**
 * 消费者
 */
public class ConsumerQuickStart {
    public static void main(String[] args) {

        // 1. kafka链接配置信息
        Properties prop = new Properties();
        // 1.1 链接地址
        prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");
        // 1.2 key value 数据序列化
        prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // 1.3 设置消费者组
        prop.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");

        // 2. 创建消费者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(prop);

        // 3. 订阅主题
        consumer.subscribe(Collections.singletonList("topic-first"));

        // 4. 拉取消息
        while (true) {
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord.key());
                System.out.println(consumerRecord.value());
            }
        }
    }
}

⑷. 测试

启动 ConsumerQuickStart、再启动ProducerQuickStart、再查看ConsumerQuickStart控制台打印结果:
在这里插入图片描述


⑸. 消费者组

在这里插入图片描述

创建 消费者2:
在这里插入图片描述

①. 一对一

生产者发送消息,多个消费者订阅同一个主题,只能有一个消费者收到消息

启动 ConsumerQuickStartConsumerQuickStart - 2、再启动ProducerQuickStart、再查看 ConsumerQuickStartConsumerQuickStart - 2控制台打印结果:

在这里插入图片描述
在这里插入图片描述


②. 一对多

生产者发送消息,多个消费者订阅同一个主题,所有消费者都能收到消息

编辑 heima-leadnews-test/kafka-demo/src/main/java/com/heima/kafka/sample/ConsumerQuickStart.java 文件:

        // 1.3 设置消费者组
        // prop.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
        prop.put(ConsumerConfig.GROUP_ID_CONFIG, "group2");

启动 ConsumerQuickStartConsumerQuickStart - 2、再启动ProducerQuickStart、再查看 ConsumerQuickStartConsumerQuickStart - 2控制台打印结果:
在这里插入图片描述
在这里插入图片描述


4. 分区机制

Kafka 中的分区机制指的是将每个主题划分成多个分区(Partition)
可以处理更多的消息,不受单台服务器的限制,可以不受限的处理更多的数据

在这里插入图片描述

分区策略:

分区策略说明
轮询策略按顺序轮流将每条数据分配到每个分区中
随机策略每次都随机地将消息分配到每个分区
按键保存策略生产者发送数据的时候,可以指定一个key,计算这个key的hashCode值,按照hashCode的值对不同消息进行存储

编辑 heima-leadnews-test/kafka-demo/src/main/java/com/heima/kafka/sample/ProducerQuickStart.java 文件:

        // 3. 发送消息
        // ProducerRecord<String, String> kvProducerRecord = new ProducerRecord<String, String>("topic-first", "key-001", "hello kafka");

        // 轮询策略(默认)
        // ProducerRecord<String, String> kvProducerRecord = new ProducerRecord<String, String>("topic-first", "hello kafka");
        // 指定分区号
        ProducerRecord<String, String> kvProducerRecord = new ProducerRecord<String, String>("topic-first", 0,"key-001", "hello kafka");

编辑 heima-leadnews-test/kafka-demo/src/main/java/com/heima/kafka/sample/ConsumerQuickStart.java 文件:

                System.out.println(consumerRecord.value());
                // 分区号
                System.out.println(consumerRecord.partition());

启动 ConsumerQuickStart 再启动ProducerQuickStart、再查看 ConsumerQuickStart `的控制台打印结果:
在这里插入图片描述


在这里插入图片描述

每一个分区都是一个顺序的、不可变的消息队列, 并且可以持续的添加。
分区中的消息都被分了一个序列号,称之为偏移量(offset),在每个分区中此偏移量都是唯一的。


5. 高可用设计

⑴. 集群

在这里插入图片描述

  • Kafka 的服务器端由被称为 Broker 的服务进程构成,即一个 Kafka 集群由多个 Broker 组成
  • 这样如果集群中某一台机器宕机,其他机器上的 Broker 也依然能够对外提供服务。这其实就是 Kafka 提供高可用的手段之一

⑵. 备份机制(Replication)

在这里插入图片描述

Kafka 中消息的备份又叫做 副本(Replica)

Kafka 定义了两类副本: 领导者副本(Leader Replica)、追随者副本(Follower Replica)

同步方式

ISR(in-sync replica)需要同步复制保存的follower

在这里插入图片描述

如果leader失效后,需要选出新的leader,选举的原则如下:

  • 选举时优先从ISR中选定,因为这个列表中follower的数据是与leader同步的
  • 如果ISR列表中的follower都不行了,就只能从其他follower中选取

极端情况,就是所有副本都失效了,这时有两种方案

  • 等待ISR中的一个活过来,选为Leader,数据可靠,但活过来的时间不确定
  • 选择第一个活过来的Replication,不一定是ISR中的,选为leader,以最快速度恢复可用性,但数据不一定完整

6. 生产者详解

⑴. 发送类型

  • 同步发送: 使用send()方法发送,它会返回一个Future对象,调用get()方法进行等待,就可以知道消息是否发送成功
  • 异步发送: 调用send()方法,并指定一个回调函数,服务器在返回响应时调用函数

编辑 heima-leadnews-test/kafka-demo/src/main/java/com/heima/kafka/sample/ProducerQuickStart.java 文件:

        //producer.send(kvProducerRecord);

        // 同步发送消息
        /*RecordMetadata recordMetadata = producer.send(kvProducerRecord).get();
        System.out.println(recordMetadata.offset());*/
        // -> 6、7、8...

        // 异步发送消息
        producer.send(kvProducerRecord, new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                if(e != null) {
                    System.out.println("记录异常信息到日志表中");
                }
                System.out.println(recordMetadata.offset());
            }
        });
        // -> 8、9、10...

⑵. 参数详解

①. 消息确认机制

在这里插入图片描述

//ack配置  消息确认机制
prop.put(ProducerConfig.ACKS_CONFIG,"all");

参数的选择说明:

确认机制说明
acks=0生产者在成功写入消息之前不会等待任何来自服务器的响应,消息有丢失的风险,但是速度最快
acks=1(默认值)只要集群首领节点收到消息,生产者就会收到一个来自服务器的成功响应
acks=all只有当所有参与赋值的节点全部收到消息时,生产者才会收到一个来自服务器的成功响应

②. 重试次数

生产者从服务器收到的错误有可能是临时性错误,在这种情况下,retries参数的值决定了生产者可以重发消息的次数,如果达到这个次数,生产者会放弃重试返回错误,默认情况下,生产者会在每次重试之间等待100ms
在这里插入图片描述

代码中配置方式:

//重试次数
prop.put(ProducerConfig.RETRIES_CONFIG,10);

③. 消息压缩

默认情况下, 消息发送时不会被压缩。

代码中配置方式:

//数据压缩
prop.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"lz4");
压缩算法说明
snappy占用较少的 CPU, 却能提供较好的性能和相当可观的压缩比, 如果看重性能和网络带宽,建议采用
lz4占用较少的 CPU, 压缩和解压缩速度较快,压缩比也很客观
gzip占用较多的 CPU,但会提供更高的压缩比,网络带宽有限,可以使用这种算法

使用压缩可以降低网络传输开销和存储开销,而这往往是向 Kafka 发送消息的瓶颈所在。


7. 消费者详解

⑴. 消费者组

由一个或多个消费者组成的群体

在这里插入图片描述

  • 一个发布在Topic上消息被分发给此消费者组中的一个消费者
    • 所有的消费者都在一个组中,那么这就变成了queue模型
    • 所有的消费者都在不同的组中,那么就完全变成了发布-订阅模型

⑵. 消息有序性

应用场景:

  • 即时消息中的单对单聊天和群聊,保证发送方消息发送顺序与接收方的顺序一致
  • 充值转账两个渠道在同一个时间进行余额变更,短信通知必须要有顺序
    在这里插入图片描述
    topic分区中消息只能由消费者组中的唯一一个消费者处理,所以消息肯定是按照先后顺序进行处理的。但是它也仅仅是保证Topic的一个分区顺序处理,不能保证跨分区的消息先后处理顺序。 所以,如果你想要顺序的处理Topic的所有消息,那就只提供一个分区。

⑶. 手动提交偏移量

kafka不会像其他JMS队列那样需要得到消费者的确认,消费者可以使用kafka来追踪消息在分区的位置(偏移量)

消费者会往一个叫做 _consumer_offset 的特殊主题发送消息,消息里包含了每个分区的偏移量。如果消费者发生崩溃或有新的消费者加入群组,就会触发再均衡
在这里插入图片描述

①. 正常的情况

在这里插入图片描述

如果消费者2挂掉以后,会发生再均衡,消费者2负责的分区会被其他消费者进行消费
再均衡后不可避免会出现一些问题

②. 问题一

在这里插入图片描述

如果提交偏移量小于客户端处理的最后一个消息的偏移量,那么处于两个偏移量之间的消息就会被重复处理。

③. 问题二:

在这里插入图片描述

如果提交的偏移量大于客户端的最后一个消息的偏移量,那么处于两个偏移量之间的消息将会丢失。
如果想要解决这些问题,还要知道目前kafka提交偏移量的方式:
提交偏移量的方式有两种,分别是自动提交偏移量和手动提交

④. 自动提交偏移量

enable.auto.commit 被设置为 true(默认),提交方式就是让消费者自动提交偏移量,每隔5秒消费者会自动把从 poll() 方法接收的最大偏移量提交上去

  • 手动提交 ,当enable.auto.commit被设置为false可以有以下三种提交方式
    • 提交当前偏移量(同步提交)
    • 异步提交
    • 同步和异步组合提交

1.提交当前偏移量(同步提交)

enable.auto.commit设置为false,让应用程序决定何时提交偏移量。使用commitSync()提交偏移量,commitSync()将会提交poll返回的最新的偏移量,所以在处理完所有记录后要确保调用了commitSync()方法。否则还是会有消息丢失的风险。

只要没有发生不可恢复的错误,commitSync()方法会一直尝试直至提交成功,如果提交失败也可以记录到错误日志里。

编辑 heima-leadnews-test/kafka-demo/src/main/java/com/heima/kafka/sample/ConsumerQuickStart.java 文件:

                // 同步提交偏移量
                try {
                    consumer.commitSync();
                }catch (Exception e) {
                    System.out.println("记录失败提交的异常: " + e);
                }

2.异步提交

手动提交有一个缺点,那就是当发起提交调用时应用会阻塞。当然我们可以减少手动提交的频率,但这个会增加消息重复的概率(和自动提交一样)。另外一个解决办法是,使用异步提交的API。

编辑 heima-leadnews-test/kafka-demo/src/main/java/com/heima/kafka/sample/ConsumerQuickStart.java 文件:

            // 异步提交偏移量
            consumer.commitAsync(new OffsetCommitCallback() {
                @Override
                public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                    if(e != null) {
                        System.out.println("记录错误的提交偏移量: " + map + ", 错误信息: " + e);
                    }
                }
            });

3.同步和异步组合提交

异步提交也有个缺点,那就是如果服务器返回提交失败,异步提交不会进行重试。相比较起来,同步提交会进行重试直到成功或者最后抛出异常给应用。异步提交没有实现重试是因为,如果同时存在多个异步提交,进行重试可能会导致位移覆盖。

举个例子,假如我们发起了一个异步提交commitA,此时的提交位移为2000,随后又发起了一个异步提交commitB且位移为3000;commitA提交失败但commitB提交成功,此时commitA进行重试并成功的话,会将实际上将已经提交的位移从3000回滚到2000,导致消息重复消费。

编辑 heima-leadnews-test/kafka-demo/src/main/java/com/heima/kafka/sample/ConsumerQuickStart.java 文件:

        // 同步和异步提交偏移量
        try {
            while (true) {
                ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                    System.out.println(consumerRecord.key());
                    System.out.println(consumerRecord.value());
                    // 分区号
                    System.out.println(consumerRecord.partition());
                    // 偏移量
                    System.out.println(consumerRecord.offset());
                }

                // 异步提交偏移量
                consumer.commitAsync();
            }
        }catch (Exception e) {
            e.printStackTrace();
            System.out.println("记录错误信息: " + e);
        }finally {
            // 同步
            consumer.commitSync();
        }




二、自媒体文章上下架

1. Spring集成kafka

⑴. Demo搭建

①. pom依赖

新建 heima-leadnews-test/kafka-demo/pom.xml 文件:

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.kafka</groupId>
                    <artifactId>kafka-clients</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
        </dependency>

②. 启动类

新建 heima-leadnews-test/kafka-demo/src/main/java/com/heima/kafka/KafkaDemoApplication.java 文件:

@SpringBootApplication
public class KafkaDemoApplication {
    public static void main(String[] args) {
        SpringApplication.run(KafkaDemoApplication.class, args);
    }
}

③. 配置信息

新建 heima-leadnews-test/kafka-demo/src/main/resources/application.yml 文件:

server:
  port: 9991
spring:
  application:
    name: kafka-demo
  kafka:
    bootstrap-servers: 192.168.200.130:9092
    producer:
      retries: 10
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: ${spring.application.name}-test
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

④. 生产者

新建 heima-leadnews-test/kafka-demo/src/main/java/com/heima/kafka/Controller/HelloController.java 文件:

/**
 * 生产者发送消息
 */
@RestController
public class HelloController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @GetMapping("/hello")
    public String hello() {
       kafkaTemplate.send("itcast-topic", "kafka 生产者发送消息");
       return "ok";
    }
}

⑤. 消费者

新建 heima-leadnews-test/kafka-demo/src/main/java/com/heima/kafka/listener/HelloListener.java 文件:

/**
 * 消费者接收消息
 */
@Component
public class HelloListener {

    @KafkaListener(topics = "itcast-topic")
    public void onMessage(String message) {
        if(!StringUtils.isEmpty(message)) {
            System.out.println(message);
        }
    }
}

⑥. 测试

启动 KafkaDemoApplication 启动类:
在这里插入图片描述
启动 http://localhost:9991/hello 地址:
在这里插入图片描述


⑵. 传递消息为对象

目前springboot整合后的kafka,因为序列化器是 StringSerializer ,这个时候如果需要传递对象可以有两种方式

  • 方式一: 可以自定义序列化器,对象类型众多,这种方式通用性不强,本章节不介绍

  • 方式二: 可以把要传递的对象进行转json字符串,接收消息后再转为对象即可,本项目采用这种方式

①. user对象

新建 heima-leadnews-test/kafka-demo/src/main/java/com/heima/kafka/pojo/User.java 文件:

/**
 * 对象
 */
public class User {

    private String userName;
    private Integer age;

    @Override
    public String toString() {
        return "User{" +
                "userName='" + userName + '\'' +
                ", age=" + age +
                '}';
    }

    public String getUserName() {
        return userName;
    }

    public void setUserName(String userName) {
        this.userName = userName;
    }

    public Integer getAge() {
        return age;
    }

    public void setAge(Integer age) {
        this.age = age;
    }
}

②. 生产者

编辑 heima-leadnews-test/kafka-demo/src/main/java/com/heima/kafka/Controller/HelloController.java 文件:

       //kafkaTemplate.send("itcast-topic", "kafka 生产者发送消息");

        // 传递消息为对象
        User user = new User();
        user.setUserName("Rebecca");
        user.setAge(6);
        kafkaTemplate.send("user-topic", JSON.toJSONString(user));

③. 消费者

编辑 heima-leadnews-test/kafka-demo/src/main/java/com/heima/kafka/listener/HelloListener.java 文件:

    // @KafkaListener(topics = "itcast-topic")
    @KafkaListener(topics = "user-topic")
    public void onMessage(String message) {
        if(!StringUtils.isEmpty(message)) {
            // System.out.println(message);

            // 传递消息为对象
            User user = JSON.parseObject(message, User.class);
            System.out.println(user);
        }
    }

④. 测试

启动 KafkaDemoApplication 启动类:
在这里插入图片描述

启动 http://localhost:9991/hello 地址:
在这里插入图片描述


2. 需求分析

  • 已发表且已上架的文章可以下架
  • 已发表且已下架的文章可以上架

流程说明:
在这里插入图片描述


3. 自媒体端(功能实现)

⑴. 接口定义

①. 说明

说明
接口路径/api/v1/news/down_or_up
请求方式POST
参数DTO
响应结果ResponseResult

②. DTO

@Data
public class WmNewsDto {
    
    private Integer id;
    /**
    * 是否上架  0 下架  1 上架
    */
    private Short enable;
                       
}

③. ResponseResult

在这里插入图片描述


⑵. 接口实现

①. DTO

编辑 heima-leadnews-model/src/main/java/com/heima/model/wemedia/dtos/WmNewsDto.java 文件:

    /**
     * 上下架 0 下架 1 上架
     */
    private Short enable;

②. Controller

编辑 heima-leadnews-service/heima-leadnews-wemedia/src/main/java/com/heima/wemedia/controller/v1/WmNewsController.java 文件:

    @PostMapping("/down_or_up")
    public ResponseResult downOrUp(@RequestBody WmNewsDto dto) {
        return wmNewsService.downOrUp(dto);
    }

③. Service

编辑 heima-leadnews-service/heima-leadnews-wemedia/src/main/java/com/heima/wemedia/service/WmNewsService.java 文件:

    /**
     * 文章上下架
     * @param dto
     * @return
     */
    public ResponseResult downOrUp(WmNewsDto dto);

④. ServiceImpl

编辑 heima-leadnews-service/heima-leadnews-wemedia/src/main/java/com/heima/wemedia/service/impl/WmNewsServiceImpl.java 文件:

    /**
     * 文章上下架
     * @param dto
     * @return
     */
    @Override
    public ResponseResult downOrUp(WmNewsDto dto) {

        // 1. 校验参数
        if(dto.getId() == null) {
            return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
        }

        // 2. 判断文章是否存在
        WmNews wmNews = getById(dto.getId());
        if(wmNews == null) {
            return ResponseResult.errorResult(AppHttpCodeEnum.DATA_NOT_EXIST, "文章不存在");
        }

        // 3. 判断文章是否已发布
        if(!wmNews.getStatus().equals(WmNews.Status.PUBLISHED.getCode())) {
            return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID, "当前文章不是发布状态, 不能上下架");
        }

        // 4. 修改文章enable(上下架状态)
        if(dto.getEnable() != null && dto.getEnable() > -1 && dto.getEnable() < 2) {
            update(Wrappers.<WmNews>lambdaUpdate().set(WmNews::getEnable, dto.getEnable())
                    .eq(WmNews::getId, wmNews.getId()));
        }
        return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);
    }

⑤. 测试

启动 ScheduleApplicationWemediaApplicationWemediaGatewayAplication,以及 Nginx,打开 自媒体端地址 http://localhost:8802 :
在这里插入图片描述


4. Kafka(传递消息)

⑴. 公共

①. 常量

新建 heima-leadnews-common/src/main/java/com/heima/common/constants/WmNewsMessageConstants.java 文件:

public class WmNewsMessageConstants {

    public static final String WM_NEWS_UP_OR_DOWN_TOPIC="wm.news.up.or.down.topic";
}

②. pom依赖

编辑 heima-leadnews-common/pom.xml 文件:

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

⑵. 发送消息

①. Nacos

编辑 leadnews-wemedia 文件:

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

在这里插入图片描述


②. 文章上下架后发送消息

编辑 heima-leadnews-service/heima-leadnews-wemedia/src/main/java/com/heima/wemedia/service/impl/WmNewsServiceImpl.java 文件:

...
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

...

        // 4. 修改文章enable(上下架状态)
        if(dto.getEnable() != null && dto.getEnable() > -1 && dto.getEnable() < 2) {
            update(Wrappers.<WmNews>lambdaUpdate().set(WmNews::getEnable, dto.getEnable())
                    .eq(WmNews::getId, wmNews.getId()));

            // kafka 发送消息, 通知Article端修改文章配置
            if(wmNews.getArticleId() != null) {
                Map<String, Object> map = new HashMap<>();
                map.put("articleId", wmNews.getArticleId());
                map.put("enable", dto.getEnable());
                kafkaTemplate.send(WmNewsMessageConstants.WM_NEWS_UP_OR_DOWN_TOPIC, JSON.toJSONString(map));
            }
        }

⑶. 接收消息

①. Nacos

编辑 leadnews-article 文件:

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

在这里插入图片描述


②. 文章上下架后接收消息

新建 heima-leadnews-service/heima-leadnews-article/src/main/java/com/heima/article/listener/ArticleIsDownListener.java 文件:

@Component
@Slf4j
public class ArticleIsDownListener {

    @KafkaListener(topics = WmNewsMessageConstants.WM_NEWS_UP_OR_DOWN_TOPIC)
    public void onMessage(String message) {
        if(!StringUtils.isNotBlank(message)) {
            Map map = JSON.parseObject(message, Map.class);
            log.info("article端文章配置修改,articleId={}",map.get("articleId"));
        }
    }
}

5. 文章端(修改文章信息)

①. Service

新建 heima-leadnews-service/heima-leadnews-article/src/main/java/com/heima/article/service/ApArticleConfigService.java 文件:

public interface ApArticleConfigService extends IService<ApArticleConfig> {

    /**
     * 修改文章配置
     */
    public void updateByMap(Map map);
}

②. ServiceImpl

编辑 heima-leadnews-service/heima-leadnews-article/src/main/java/com/heima/article/service/impl/ApArticleConfigServiceImpl.java 文件:

@Service
@Slf4j
@Transactional
public class ApArticleConfigServiceImpl extends ServiceImpl<ApArticleConfigMapper, ApArticleConfig> implements ApArticleConfigService {

    /**
     * 修改文章配置
     */
    @Override
    public void updateByMap(Map map) {

        // 0 下架 1 上架
        Object enable = map.get("enable");
        boolean isDown = true;
        if(enable.equals(1)) {
            isDown = false;
        }

        // 修改文章配置
        update(Wrappers.<ApArticleConfig>lambdaUpdate().eq(ApArticleConfig::getArticleId, map.get("articleId"))
                .set(ApArticleConfig::getIsDown, isDown));
    }
}

③. 修改文章配置

编辑 heima-leadnews-service/heima-leadnews-article/src/main/java/com/heima/article/listener/ArticleIsDownListener.java 文件:

    @Autowired
    private ApArticleConfigService apArticleConfigService;

...

            Map map = JSON.parseObject(message, Map.class);
            // 监听, 修改文章配置
            apArticleConfigService.updateByMap(map);


④. 测试

启动 ScheduleApplicationArticleApplicationWemediaApplicationWemediaGatewayAplication,以及 Nginx,打开 自媒体端地址 http://localhost:8802 :

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述



本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/188960.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

并查集应用

一、并查集模板 int find(int x) {if(p[x]!x) p[x]find(p[x]);return p[x]; }并查集高效率的核心是一旦更新过一次后&#xff0c;就会将路径压缩掉&#xff0c;避免后续重复遍历路径。 二、并查集应用 1、格子游戏 分析&#xff1a;每构成一个方框&#xff0c;当最后两个点连…

RA4M2开发(2)----基于IIC驱动OLED

概述 在e2studio中创建新的工程并导入必要的文件&#xff0c;包括I2C驱动代码和SSD1306 OLED显示驱动代码。配置RA4M2的I2C接口&#xff0c;使其作为I2C master进行通信。初始化SSD1306 OLED显示驱动代码&#xff0c;并配置显示屏的物理地址和分辨率。通过I2C驱动代码将数据写…

【Linux】初识环境变量

文章目录环境变量引入初见环境变量和环境变量有关的指令如何通过代码获取环境变量getenv()main函数的命令行参数第三方变量environ程序变量可以继承给子进程环境变量引入 Linux中有各种指令&#xff0c; 每个指令其实都是一个可执行程序&#xff1a; 和我们自己写的C语言代码…

API自动化测试【postman生成报告】

PostMan生成测试报告有两种&#xff1a; 1、控制台的模式 2、HTML的测试报告 使用到一个工具newman Node.js是前端的一个组件&#xff0c;主要可以使用它来开发异步的程序。 一、控制台的模式 1、安装node.js 双击node.js进行安装&#xff0c;安装成功后在控制台输入node…

Ansys Zemax | 多模光纤耦合

本文展示了利用几何图像分析特性来计算多模光纤耦合效率的方法。 还有使用IMAE操作数优化多模光纤耦合效率的方法。该方法只适用于包含大量模式的多模光纤。 下载 联系工作人员获取附件 简介 我们可以使用OpticStudio中的几何图像分析&#xff08;Geometric Image Analysi…

已解决error: legacy-instal1-failure

已解决&#xff08;pip install wxPython安装失败&#xff09;error: legacy-instal1-failure Encountered error while trying to install package.wxPython note: This is an issue with the package mentioned above&#xff0c;not pip. hint : See above for output from …

ChatGPT 回答什么是敏捷测试

Jenkins 官方文档&#xff1a;https://jenkins.io/doc/JUnit 插件文档&#xff1a;JUnit | Jenkins pluginTestNG 插件文档&#xff1a;https://plugins.jenkins.io/testng-plugin/Jenkins 教程&#xff1a;https://jenkins.io/doc/tutorials/JUnit 教程&#xff1a;JUnit 5 Us…

舆情监测软件报价方案,TOOM舆情监控软件多少钱?

舆情监测软件的价格因公司不同而有所差异&#xff0c;通常在几百元至数千元不等。价格还可能受到功能复杂度、数据量、使用时长等因素的影响。建议您联系多家软件开发商或销售代表&#xff0c;对比各家的产品和报价&#xff0c;以找到适合您需求和预算的解决方案。舆情监测软件…

算法训练营DAY46|139.单词拆分、多重背包的介绍

这一期到了背包问题的最后一期&#xff0c;主要讲解一道leetcodee题&#xff0c;和对多重背包的一些简单介绍&#xff0c;由于leetcode没有对于多重背包的具体问题&#xff0c;且面试基本不会问到多重背包所以只是作为科普。 139. 单词拆分 - 力扣&#xff08;LeetCode&#x…

缺省参数,函数重载,引用的使用(C++)

前言 本文主要讲解的是缺省参数&#xff0c;函数重载&#xff0c;引用的使用. 文章目录 目录 前言 一、缺省参数是什么&#xff1f; 1.缺省参数的分类 二、函数重载是什么&#xff1f; 三、引用是什么&#xff1f; 1.引用特性 总结 提示&#xff1a;以下是本篇文章正文内容&am…

java中的NIO,BIO,AIO

IO&#xff0c;input和output&#xff0c;Java 为 I/O 提供了强大的而灵活的支持&#xff0c;使其更广泛地应用到文件传输和网络编程中。 IO是什么&#xff1f;为什么需要IO&#xff1f; java在1.4之前&#xff0c;提供的是BIO&#xff0c;也就是java.io包下的的东西&#xf…

SpringBoot(二):配置文件的作用、配置文件的格式、properties配置文件、yml配置文件

目录一、配置文件的作用二、配置文件的格式三、properties配置文件3.1 properties的基本语法3.2 properties的缺点3.3 配置自定义内容3.4 读取配置文件四、yml配置文件4.1 yml基本介绍4.2 yml基本语法4.3 使用yml配置不同的数据类型4.4 读取yml配置文件4.5 在yml中配置对象4.6 …

关于UWB汽车钥匙介绍

汽车钥匙经历了机械钥匙、遥控钥匙、PEPS、数字钥匙四个阶段&#xff0c;而数字钥匙又分为BLE/NFC/UWB三种技术路线。由于UWB安全性、定位精度、作用范围明显好于BLE和NFC&#xff0c;因此成为汽车数字钥匙的最优技术。PEPS与数字钥匙PEPS是指无钥匙进入/无钥匙启动系统&#x…

MySQL详解(三)——高级 1.0

MySQL高级 utf8只支持每个字符最多三个字节&#xff0c;而真正的 UTF-8 是每个字符最多四个字节 MySQL 的这个bug一直没有被修复&#xff0c;他们在 2010 年发布了一个叫作“utf8mb4”的字符集&#xff0c;绕过了这个问题。 MySQL优点 Mysql是开源的Mysql是可以定制的&…

CDN的那些事

CDN的那些事一、前言二、基础配置三、访问控制四、缓存配置五、HTTPS配置六、高级配置&#xff08;最重点来了&#xff09;七、结语一、前言 大家好&#xff0c;我是尝试中成长的站长&#xff0c;前段时间&#xff0c;群友的cdn被刷爆了&#xff0c;这就引起了我的恐慌&#x…

读书笔记:神经网络加法层与乘法层的实现及应用 buy_apple_orange.py ← 斋藤康毅

“层”是神经网络中功能的单位。通常把神经网络的“层”实现为一个类。“层”的实现中有两个共通的方法 forward() 和 backward()。forward() 对应正向传播&#xff0c;backward() 对应反向传播。加法结点的反向传播将上游的值原封不动地输出到下游&#xff0c;乘法结点的反向传…

第2章:使用CSS定义样式

在前一章中,我们研究了Java代码的不同片段。 在本章中,我们将对每个示例使用相同的代码段: public void createPdf(String html, String dest) throws IOException {HtmlConverter.convertToPdf(html, new FileOutputStream(dest)); } 我们不看Java代码的不同片段,而是看…

数据分析-深度学习Pytorch Day13

单层感知机和多层感知机(MLP)是最基础的神经网络结构。将卷积操作创新的加入到神经网络结构形成了卷积神经网络&#xff0c;卷积神经网络给现代人工智能注入了活力。感知机网络和卷积网络(CNN)都属于前馈型网络(FeedForward Network)。单层感知机是二分类的线性分类模型&#x…

MT1010-M1016(java版)

MT1010输入和输出字符型数据难度&#xff1a;青铜0时间限制&#xff1a;1秒巴占用内存&#xff1a;64M★收藏△报错请编写一个简单程序&#xff0c;用户输入2个的字符型数据存储在变量中&#xff0c;并分别以字符形式和整数形式输出在屏幕上。格式输入格式&#xff1a;2个的字符…

Go语言DDD实战初级篇

导读 领域驱动设计(DDD)最简洁的描述可能是&#xff1a;如何在明确的限界上下文中创建通用语言的模型。通过 DDD思想设计开发的软件&#xff0c;在领域专家、开发者和软件本身之间不存在“翻译”&#xff0c;三者通过在限界上下文下的通用语言直接表示。而这个系列则是我们团队…