【2024】kafka streams结合案例分析举行实际项目开辟使用（3） ...

不到断气不罢休  金牌会员 | 2024-10-24 07:07:34 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 876|帖子 876|积分 2628

一、前沿先容

前面已经大致先容了kafka streams的基本使用了，这里结合一个实际案例来举行训练使用kafka streams。
下面案例是一个阛阓购物的场景，就比如我们去一个购物阛阓购买东西时，在购买的时候。阛阓会记录下来我们这一次消耗的信息，一样平常首先会先把银行卡等信息举行一个加***隐蔽，然后再把信息分别发送发送给需要的topic，如累计积分的，把购买的金额转为积分返回给用户账号；根据购买产品的不同发送给不同的topic。具体如下：
   实行流程
  

  • 通过split把用户购买产品的记录存入到不同的分支

    • 咖啡写入caffee 处置处罚器
    • 电子产品写入electronics处置处罚器

  • 把支付的金额以积分的形式传入到pattem处置处罚器
  • 把transactionKey作为key，value为原始数据传入到purchase处置处罚器
  • 把支付的金额以积分的形式传入到reward处置处罚器
  • 再把原始数据全部写到data数据仓库去
  

使用到的实体类

二、代码实现

1、依靠

和前面类似，重要是kafka的相关依靠
  1.     <dependency>
  2.       <groupId>org.apache.kafka</groupId>
  3.       <artifactId>kafka-clients</artifactId>
  4.     </dependency>
  6.     <dependency>
  7.       <groupId>org.apache.kafka</groupId>
  8.       <artifactId>kafka-streams</artifactId>
  9.     </dependency>
  10.         <dependency>
  11.       <groupId>cn.hutool</groupId>
  12.       <artifactId>hutool-all</artifactId>
  13.       <version>5.8.20</version>
  14.     </dependency>
  15.     <dependency>
  16.       <groupId>org.projectlombok</groupId>
  17.       <artifactId>lombok</artifactId>
  18.       <optional>true</optional>
  19.     </dependency>
  20.         <dependency>
  21.       <groupId>com.google.code.gson</groupId>
  22.       <artifactId>gson</artifactId>
  23.       <version>2.8.7</version>
  24.     </dependency>
复制代码
2、实体类

2.1、Transaction

写入的购买信息
  1. @Data
  2. @Builder
  3. public class Transaction {
  4.     /**性*/
  5.     private String lastName;
  6.     /**名*/
  7.     private String firstName;
  8.     /**顾客id*/
  9.     private String customerId;
  10.     /**银行卡号*/
  11.     private String creditCardNumber;
  12.     /**商品名称*/
  13.     private String itemPurchased;
  14.     /**店铺名称*/
  15.     private String department;
  16.     /**数量*/
  17.     private Integer quantity;
  18.     /**价格*/
  19.     private Double price;
  20.     /**购买时间*/
  21.     private String purchaseDate;
  22.     /**邮政编码*/
  23.     private String zipCode;
  24. }
复制代码
2.2、 TransactionKey

用作标记，转为key的
  1. @Data
  2. @Builder
  3. public class TransactionKey {
  5.     private String customerId;
  7.     private String purchaseDate;
  8. }
复制代码
2.3、TransactionPattern

记录购买信息
  1. @Data
  2. @Builder
  3. public class TransactionPattern {
  5.     private String zipCode;
  7.     private String item;
  9.     private String date;
  11.     private Double amount;
  12. }
复制代码
2.4、CustomerReward

  1. @Data
  2. @Builder
  3. public class CustomerReward {
  5.     private String customerId;
  7.     private String purchaseTotal;
  9.     private Integer rewardPoints;
  10. }
复制代码
3、序列化工具类

3.1、序列化

  1. /**
  2. * 序列化
  3. * @param <T>
  4. */
  5. public class JsonSerializer<T> implements Serializer<T> {
  7.     private Gson gson=  new Gson();
  11.     public void configure(Map<String ,?> map, boolean b) {
  12.     }
  16.     public byte[] serialize(String topic, T t) {
  19.         return gson.toJson(t).getBytes();
  20.     }
  24.     @Override
  25.     public void close() {
  26.         Serializer.super.close();
  27.     }
  28. }
复制代码
3.2、反序列化

  1. /**
  2. * 反序列化
  3. * @param <T>
  4. */
  5. public class JsonDeserializer<T> implements Deserializer<T> {
  6.     private Gson gson=  new Gson();
  8.     private Class<T> deserializeClass;
  9.     public JsonDeserializer(Class<T> deserializeClass){
  10.         this.deserializeClass=deserializeClass;
  11.     }
  13.     public JsonDeserializer(){
  15.     }
  17.     @Override
  18.     @SuppressWarnings("unchecked")
  19.     public void configure(Map<String,?> map, boolean b){
  20.         if (deserializeClass == null){
  21.             deserializeClass = (Class<T>) map.get("serializedClass");
  22.         }
  23.     }
  25.     @Override
  26.     public T deserialize(String topic, byte[] data) {
  27.         if (data == null){
  28.             return null;
  29.         }
  30.         return gson.fromJson(new String(data),deserializeClass);
  31.     }
  34.     @Override
  35.     public void close() {
  37.     }
  38. }
复制代码
3.3、Serde仓库

用做直接通过调用实现Serde使用json序列化转换，也可以参考Serdes方法实现
  1. /**
  2. * 序列化和反序列化
  3. */
  4. public class JsonSerdes {
  6.     /**获取Serde*/
  7.     public static TransactionPatternWrapSerde TransactionPattern() {
  8.         return new TransactionPatternWrapSerde(new JsonSerializer<>(), new JsonDeserializer<>(TransactionPattern.class));
  9.     }
  10.     public static TransactionKeyWrapSerde TransactionKey() {
  11.         return new TransactionKeyWrapSerde(new JsonSerializer<>(), new JsonDeserializer<>(TransactionKey.class));
  12.     }
  13.     public static CustomerRewardWrapSerde CustomerReward() {
  14.         return new CustomerRewardWrapSerde(new JsonSerializer<>(), new JsonDeserializer<>(CustomerReward.class));
  15.     }
  17.     public static TransactionWrapSerde Transaction() {
  18.         return new TransactionWrapSerde(new JsonSerializer<>(), new JsonDeserializer<>(Transaction.class));
  19.     }
  25.     /**创建Serde*/
  27.     private final static class TransactionPatternWrapSerde extends WrapSerde<TransactionPattern>{
  29.         public TransactionPatternWrapSerde(Serializer<TransactionPattern> serializer, Deserializer<TransactionPattern> deserializer) {
  30.             super(serializer, deserializer);
  31.         }
  32.     }
  35.     private final static class TransactionKeyWrapSerde extends WrapSerde<TransactionKey>{
  37.         public TransactionKeyWrapSerde(Serializer<TransactionKey> serializer, Deserializer<TransactionKey> deserializer) {
  38.             super(serializer, deserializer);
  39.         }
  40.     }
  42.     private final static class CustomerRewardWrapSerde extends WrapSerde<CustomerReward>{
  43.         public CustomerRewardWrapSerde(Serializer<CustomerReward> serializer, Deserializer<CustomerReward> deserializer) {
  44.             super(serializer, deserializer);
  45.         }
  46.     }
  49.     private final static class TransactionWrapSerde extends WrapSerde<Transaction>{
  51.         public TransactionWrapSerde(Serializer<Transaction> serializer, Deserializer<Transaction> deserializer) {
  52.             super(serializer, deserializer);
  53.         }
  54.     }
  57.     /** WrapSerde父类*/
  59.     private static class WrapSerde<T> implements Serde<T>{
  60.         private final Serializer<T> serializer;
  61.         private final Deserializer<T> deserializer;
  63.         public WrapSerde(Serializer<T> serializer, Deserializer<T> deserializer) {
  64.             this.serializer = serializer;
  65.             this.deserializer = deserializer;
  66.         }
  68.         @Override
  69.         public Serializer<T> serializer() {
  70.             return serializer;
  71.         }
  73.         @Override
  74.         public Deserializer<T> deserializer() {
  75.             return deserializer;
  76.         }
  77.     }
  78. }
复制代码
4、具体streams实现

使用上比力简单，重要是通过前面学的方法举行实现不同的处置处罚器转换数据，然后在发送到不同的topic中去，编写好之后，我们需要创建需要使用到的topic
  1. @Slf4j
  2. public class ShoppingStreams {
  5.     private static final String BOOTSTRAP_SERVERS = "localhost:9092";
  7.     private static final String APPLICATION_ID = "shopping-streams";
  9.     private static final String SELL_TRANSACTION_SOURCE_TOPIC = "sell.transaction";
  11.     private static final String SELL_TRANSACTION_PATTERN_TOPIC = "sell.pattern.transaction";
  13.     private static final String SELL_TRANSACTION_REWARDS_TOPIC = "sell.rewards.transaction";
  15.     private static final String SELL_TRANSACTION_COFFEE_TOPIC = "sell.coffee.transaction";
  17.     private static final String SELL_TRANSACTION_ELECT_TOPIC = "sell.elect.transaction";
  19.     private static final String SELL_TRANSACTION_PURCHASE_TOPIC = "sell.purchase.transaction";
  23.     public static void main(String[] args) throws InterruptedException {
  24.         Properties properties = new Properties();
  25.         properties.put(StreamsConfig.APPLICATION_ID_CONFIG, APPLICATION_ID);
  26.         properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
  27.         properties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, "org.apache.kafka.streams.errors.LogAndContinueExceptionHandler");
  29.         StreamsConfig streamsConfig = new StreamsConfig(properties);
  31.         Serde<String> stringSerde = Serdes.String();
  34.         StreamsBuilder builder = new StreamsBuilder();
  36. //        1、到topic中去读取数据
  37.         KStream<String, Transaction> k0 = builder
  38.                 .stream(SELL_TRANSACTION_SOURCE_TOPIC, Consumed.with(stringSerde, JsonSerdes.Transaction())
  39.                         .withName("transaction-source")
  40.                         .withOffsetResetPolicy(Topology.AutoOffsetReset.LATEST)); //指定偏移重置策略。当没有初始偏移量或偏移量超出范围时，消费将从最新的记录开始。
  42. //       2、把原始数据进行加密
  43.         KStream<String, Transaction> k1 = k0.peek((k, v) -> log.info("k:{},v:{}", k, v))
  44.                 .mapValues(v -> {
  45.                     String encryption = v.getCreditCardNumber().replaceAll("(?<=^.{4}).*", "****");
  46.                     v.setCreditCardNumber(encryption);
  47.                     return v;
  48.                 }, Named.as("pattern-sink"));
  50. //         2、记录商品购买
  51.         k1.mapValues(v -> TransactionPattern.builder()
  52.                                         .zipCode(v.getZipCode())
  53.                                         .item(v.getItemPurchased())
  54.                                         .date(v.getPurchaseDate().toString())
  55.                                         .amount(v.getPrice())
  56.                                         .build()
  57.                         , Named.as("transaction-pattern"))
  58.                 .to(SELL_TRANSACTION_PATTERN_TOPIC, Produced.with(stringSerde, JsonSerdes.TransactionPattern()));
  60. //        3、奖励用户积分
  61.         k1.mapValues(v -> CustomerReward.builder()
  62.                                 .customerId(v.getCustomerId())
  63.                                 .purchaseTotal(v.getItemPurchased())
  64.                                 .rewardPoints(v.getPrice().intValue())
  65.                                 .build()
  66.                     , Named.as("transaction-rewards"))
  67.                 .to(SELL_TRANSACTION_REWARDS_TOPIC, Produced.with(stringSerde, JsonSerdes.CustomerReward()));
  69. //        4、把消费金额大于5的记录下来（标注为key，发送出去）
  70.         k1.filter((k, v) -> v.getPrice() > 5)
  71.                 .selectKey((k, v) -> TransactionKey.builder()
  72.                                         .customerId(v.getCustomerId())
  73.                                         .purchaseDate(v.getPurchaseDate())
  74.                                         .build()
  75.                     , Named.as("transaction-purchase"))
  76.                 .to(SELL_TRANSACTION_PURCHASE_TOPIC, Produced.with(JsonSerdes.TransactionKey(), JsonSerdes.Transaction()));
  78. //        5、把购买的商品根据类型分别发送到不同的topic中
  79.         k1.split(Named.as("branch-"))
  80.                 .branch((k, v) -> v.getItemPurchased().equalsIgnoreCase("coffee")
  81.                         , Branched.withConsumer(ks->ks.to(SELL_TRANSACTION_COFFEE_TOPIC, Produced.with(stringSerde, JsonSerdes.Transaction()))))
  82.                 .branch((k, v) -> v.getItemPurchased().equalsIgnoreCase("elect")
  83.                         , Branched.withConsumer(ks->ks.to(SELL_TRANSACTION_ELECT_TOPIC, Produced.with(stringSerde, JsonSerdes.Transaction()))))
  84.                 .noDefaultBranch();
  87. //        模拟把数据全部写入到数据仓库
  88.         k1.print(Printed.<String,Transaction>toSysOut().withName("DW"));
  90.         k1.foreach((k, v) -> log.info("数据存入数据仓库=========>,k:{},v:{}", k, v));
  92.         KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), streamsConfig);
  94.         CountDownLatch latch = new CountDownLatch(1);
  95.         Runtime.getRuntime().addShutdownHook(new Thread(() -> {
  96.             kafkaStreams.close();
  97.             latch.countDown();
  98.             log.info("The Kafka Streams 执行关闭！");
  99.         }));
  101.         kafkaStreams.start();
  102.         log.info("kafka streams 启动成功！>>>>");
  103.         latch.await();
  105.     }
  106. }
复制代码
5、其他测试使用

5.1、生产者

创建一个生产消息的类，往topic发送消息
  1. /**
  2. * 生产购物消息到kafka
  3. */
  4. @Slf4j
  5. public class ShoppingProducer {
  7.     private final static String TOPIC_NAME = "sell.transaction";
  9.     private static final String BOOTSTRAP_SERVERS = "localhost:9092";
  12.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  13.         Properties props = new Properties();
  14. //        设置参数
  15.         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,BOOTSTRAP_SERVERS);
  16. //        设置序列化
  17.         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  18.         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  20. //        连接客户端
  21.         KafkaProducer<String, String> producer = new KafkaProducer<>(props);
  22.         //        发送的消息记录器(topic,partition(指定发到哪个),key（用于计算发到哪个partition）,value)
  24.         for (int i = 0; i < 5; i++) {
  25.             Transaction transaction = Transaction.builder()
  26.                     .customerId("011223")
  27.                     .itemPurchased("elect")
  28.                     .quantity(i)
  29.                     .zipCode("10100")
  30.                     .firstName("李")
  31.                     .lastName("四")
  32.                     .price(i * 100.0)
  33.                     .purchaseDate(new Date().toString())
  34.                     .creditCardNumber("4322-1223-1123-000" + i)
  35.                     .department("体育西路35号")
  36.                     .build();
  38.             String json = new JSONObject(transaction).toString();
  39. //          默认partition数量和Broker创建的数量一致
  40.             ProducerRecord<String, String> producerRecord = new ProducerRecord<>(TOPIC_NAME, 0,"my-keyValue3", json);
  42. //        同步
  43.             send(producer,producerRecord);
  44.         }
  48.     }
  50.     /**
  51.      * @param producer: 客户端对象
  52.      * @return void
  53.      * 同步发送
  54.      * @date 2024/3/22 17:09
  55.      */
  56.     private static void send(KafkaProducer<String, String> producer,ProducerRecord<String, String> producerRecord) throws InterruptedException, ExecutionException {
  58. //          等待发送成功的阻塞方法
  59.         RecordMetadata metadata = producer.send(producerRecord).get();
  61.         log.info("同步发送消息"+ "topic-"+metadata.topic()+"====partition："+metadata.partition()
  62.                 +"=====offset："+metadata.offset());
  63.     }
  66. }
复制代码
5.2、日志文件

由于kafka一直会刷日志，以是需要有一个日志文件屏蔽debug类型的日志输出
在resources路径下创建一个logback.xml文件
  1. <configuration scon="true" scanPeriod="10 seconds">
  2.     <include resource="org/springframework/boot/logging/logback/base.xml"/>
  3.     <!-- 屏蔽kafka debug -->
  4.     <logger name="org.apache.kafka.clients" level="ERROR" />
  6. </configuration>
复制代码
6、创建topic

首先需要有自己的kafka，如何创建可以看我前面的文章

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

不到断气不罢休

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

存储 挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表