将Spring Cloud Stream v2迁移到Spring Cloud Stream v4 for org.springframework.cloud.stream.messaging

ocebsuys  于 2024-01-05  发布在  Spring
关注(0)|答案(1)|浏览(133)

简介

我目前正在使用来自Spring-cloud-stream v2的Sink接口：https://javadoc.io/doc/org.springframework.cloud/spring-cloud-stream/2.0.0.RELEASE/index.html
我写了单元测试来测试我的代码。

我的项目在迁移前使用这些注解的代码：

StreamUtils.java

  1. package com.enterprise.production.event.consumer;
  3. import com.enterprise.production.model.message.EnergyMeasure;
  4. import com.enterprise.production.model.message.EnergyProductionMessage;
  5. import com.enterprise.production.model.message.Units;
  6. import com.enterprise.production.model.time.InstantPeriod;
  7. import com.enterprise.production.repository.entity.ProductionEntity;
  8. import com.enterprise.production.stream.service.ProductionMessageConverter;
  9. import com.fasterxml.jackson.databind.ObjectMapper;
  10. import com.google.common.collect.ImmutableMap;
  11. import lombok.extern.slf4j.Slf4j;
  12. import org.springframework.beans.factory.annotation.Autowired;
  13. import org.springframework.cloud.stream.messaging.Sink;
  14. import org.springframework.messaging.Message;
  15. import org.springframework.messaging.support.MessageBuilder;
  16. import org.springframework.stereotype.Component;
  18. import java.io.IOException;
  19. import java.time.Instant;
  20. import java.time.ZoneId;
  21. import java.time.temporal.ChronoUnit;
  22. import java.util.ArrayList;
  23. import java.util.Arrays;
  24. import java.util.List;
  25. import java.util.Map;
  26. import java.util.stream.Collectors;
  27. import java.util.stream.IntStream;
  29. import static com.enterprise.production.event.consumer.ProductionMessageConsumerIT.NOW;
  31. @Component
  32. @Slf4j
  33. public class StreamUtils {
  35.     @Autowired
  36.     private Sink sink;
  38.     @Autowired
  39.     ProductionMessageConverter productionMessageConverter;
  41.     @Autowired
  42.     ObjectMapper objectMapper;
  44.     public List<ProductionEntity> generateAndSend2MessagesWith2Productions(int start, int end, String deviceId, String username, String tenantId, String unitsValue, String startDate) {
  45.         List<ProductionEntity> result = new ArrayList<>();
  46.         IntStream.rangeClosed(start, end)
  47.                 .boxed()
  48.                 .collect(Collectors.toList())
  49.                 .stream()
  50.                 .forEach(i -> {
  51.                     int dateInc = i * 4;
  52.                     Instant period1From = Instant.parse(startDate).plus(dateInc, ChronoUnit.DAYS);
  53.                     Instant period1To = Instant.parse(startDate).plus(dateInc + 1, ChronoUnit.DAYS);
  54.                     Instant period2From = Instant.parse(startDate).plus(dateInc + 2, ChronoUnit.DAYS);
  55.                     Instant period2To = Instant.parse(startDate).plus(dateInc + 3, ChronoUnit.DAYS);
  56.                     Instant expiry = Instant.parse("2019-01-01T00:00:00.000Z");
  57. //                    dateInc = dateInc + 4;
  59.                     EnergyMeasure production1 = EnergyMeasure.builder()
  60.                                                              .period(new InstantPeriod(period1From, period1To))
  61.                                                              .expiry(expiry)
  62.                                                              .production(1.6)
  63.                                                              .networkConsumption(8.0)
  64.                                                              .selfConsumption(1.6)
  65.                                                              .build();
  66.                     EnergyMeasure production2 = EnergyMeasure.builder()
  67.                                                              .period(new InstantPeriod(period2From, period2To))
  68.                                                              .expiry(expiry)
  69.                                                              .production(1.6)
  70.                                                              .networkConsumption(0.0)
  71.                                                              .selfConsumption(1.6)
  72.                                                              .build();
  74.                     Map<String, Units> units = ImmutableMap.<String, Units>builder()
  75.                             .put("production", new Units(unitsValue))
  76.                             .put("selfConsumption", new Units(unitsValue))
  77.                             .put("networkConsumption", new Units(unitsValue))
  78.                             .build();
  80.                     EnergyProductionMessage message = EnergyProductionMessage.builder()
  81.                                                      .deviceId(deviceId)
  82.                                                      .username(username)
  83.                                                      .tenantId(tenantId)
  84.                                                      .units(units)
  85.                                                      .solarEnergies(Arrays.asList(production1, production2))
  86.                                                      .build();
  88.                     List<ProductionEntity> productions = productionMessageConverter.convert(message);
  89.                     productions.forEach(productionEntity -> productionEntity.setExpiry(computeTtl(productionEntity.getBeginTs())));
  90.                     result.addAll(productions);
  91.                     sendMessage(message);
  92.                 });
  93.         return result;
  94.     }
  96.     private long computeTtl(Instant fromBeginTs) {
  97.         return fromBeginTs.atZone(ZoneId.of("UTC")).plusMonths(25).toEpochSecond()
  98.                 - NOW.toEpochMilli() / 1000;
  99.     }
  101.     public void sendMessage(EnergyProductionMessage message)  {
  102.         sink.input().send(getMessage(message));
  103.     }
  105.     public Message<EnergyProductionMessage> getMessage(EnergyProductionMessage message){
  106.         return  MessageBuilder.withPayload(message).build();
  107.     }
  109.     public void sendMessage(String payload) throws IOException {
  110.         EnergyProductionMessage energyProductionMessage = objectMapper.readValue(payload,
  111.                                                                                  EnergyProductionMessage.class);
  112.         Message<EnergyProductionMessage> message = MessageBuilder.withPayload(energyProductionMessage).build();
  113.         sink.input().send(message);
  114.     }
  115. }

字符串

ProductionMessageConsumerIT.java

  1. package com.enterprise.production.event.consumer;
  3. import com.enterprise.production.Application;
  4. import com.enterprise.production.client.ClientException;
  5. import com.enterprise.production.model.message.EnergyMeasure;
  6. import com.enterprise.production.model.message.EnergyProductionMessage;
  7. import com.enterprise.production.model.message.Units;
  8. import com.enterprise.production.model.time.InstantPeriod;
  9. import com.enterprise.production.repository.ProductionRepository;
  10. import com.enterprise.production.repository.entity.ProductionEntity;
  11. import com.google.common.collect.ImmutableMap;
  12. import org.apache.commons.lang3.builder.EqualsBuilder;
  13. import org.json.JSONArray;
  14. import org.json.JSONException;
  15. import org.json.JSONObject;
  16. import org.junit.Test;
  17. import org.junit.jupiter.api.Assertions;
  18. import org.junit.runner.RunWith;
  19. import org.springframework.beans.factory.annotation.Autowired;
  20. import org.springframework.boot.test.context.SpringBootTest;
  21. import org.springframework.cloud.contract.wiremock.AutoConfigureWireMock;
  22. import org.springframework.context.annotation.Bean;
  23. import org.springframework.context.annotation.Configuration;
  24. import org.springframework.context.annotation.Primary;
  25. import org.springframework.core.io.Resource;
  26. import org.springframework.core.io.ResourceLoader;
  27. import org.springframework.test.context.ActiveProfiles;
  28. import org.springframework.test.context.junit4.SpringRunner;
  30. import java.io.IOException;
  31. import java.nio.file.Files;
  32. import java.time.Clock;
  33. import java.time.Instant;
  34. import java.time.ZoneId;
  35. import java.time.ZoneOffset;
  36. import java.time.temporal.ChronoUnit;
  37. import java.util.*;
  38. import java.util.concurrent.TimeUnit;
  39. import java.util.stream.Stream;
  41. import static com.github.tomakehurst.wiremock.client.WireMock.*;
  43. @RunWith(SpringRunner.class)
  44. @SpringBootTest(classes = {Application.class, ProductionMessageConsumerIT.TestConfig.class})
  45. @AutoConfigureWireMock(port = 0)
  46. @ActiveProfiles("it")
  47. public class ProductionMessageConsumerIT {
  49.     public static final Instant NOW = Instant.parse("2018-07-15T15:15:21Z");
  51.     @Autowired
  52.     ProductionRepository productionRepository;
  54.     @Autowired
  55.     StreamUtils streamUtils;
  57.     @Autowired
  58.     ResourceLoader resourceLoader;
  60.     @Configuration
  61.     static class TestConfig {
  62.         @Bean
  63.         @Primary
  64.         public Clock clockIT() {
  65.             return Clock.fixed(NOW, ZoneOffset.UTC); // timestamp 1531667721
  66.         }
  67.     }
  69.     public static final String CAS_TABLE_ENERGIES = "energies_v2";
  70.     public static final String CAS_TABLE_ENERGIES30m = "energies30m_v2";
  72.     // Some tests ...
  73. }

型

我的项目在迁移后使用这些注解的代码：

StreamUtils.java

  1. package com.enterprise.production.event.consumer;
  3. import com.enterprise.production.model.message.EnergyMeasure;
  4. import com.enterprise.production.model.message.EnergyProductionMessage;
  5. import com.enterprise.production.model.message.Units;
  6. import com.enterprise.production.model.time.InstantPeriod;
  7. import com.enterprise.production.repository.entity.ProductionEntity;
  8. import com.enterprise.production.stream.service.ProductionMessageConverter;
  9. import com.fasterxml.jackson.databind.ObjectMapper;
  10. import com.google.common.collect.ImmutableMap;
  11. import lombok.extern.slf4j.Slf4j;
  12. import org.springframework.beans.factory.annotation.Autowired;
  13. import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel;
  14. import org.springframework.messaging.Message;
  15. import org.springframework.messaging.support.MessageBuilder;
  16. import org.springframework.stereotype.Component;
  18. import java.io.IOException;
  19. import java.time.Instant;
  20. import java.time.ZoneId;
  21. import java.time.temporal.ChronoUnit;
  22. import java.util.ArrayList;
  23. import java.util.Arrays;
  24. import java.util.List;
  25. import java.util.Map;
  26. import java.util.stream.Collectors;
  27. import java.util.stream.IntStream;
  29. import static com.enterprise.production.event.consumer.ProductionMessageConsumerIT.NOW;
  31. @Component
  32. @Slf4j
  33. public class StreamUtils {
  35.     @Autowired
  36.     ProductionMessageConverter productionMessageConverter;
  37.     @Autowired
  38.     ObjectMapper objectMapper;
  40.     @Autowired
  41.     private DirectWithAttributesChannel sink;
  43.     public List<ProductionEntity> generateAndSend2MessagesWith2Productions(int start, int end, String deviceId, String username, String tenantId, String unitsValue, String startDate) {
  44.         List<ProductionEntity> result = new ArrayList<>();
  45.         IntStream.rangeClosed(start, end)
  46.                 .boxed()
  47.                 .collect(Collectors.toList())
  48.                 .stream()
  49.                 .forEach(i -> {
  50.                     int dateInc = i * 4;
  51.                     Instant period1From = Instant.parse(startDate).plus(dateInc, ChronoUnit.DAYS);
  52.                     Instant period1To = Instant.parse(startDate).plus(dateInc + 1, ChronoUnit.DAYS);
  53.                     Instant period2From = Instant.parse(startDate).plus(dateInc + 2, ChronoUnit.DAYS);
  54.                     Instant period2To = Instant.parse(startDate).plus(dateInc + 3, ChronoUnit.DAYS);
  55.                     Instant expiry = Instant.parse("2019-01-01T00:00:00.000Z");
  56. //                    dateInc = dateInc + 4;
  58.                     EnergyMeasure production1 = EnergyMeasure.builder()
  59.                             .period(new InstantPeriod(period1From, period1To))
  60.                             .expiry(expiry)
  61.                             .production(1.6)
  62.                             .networkConsumption(8.0)
  63.                             .selfConsumption(1.6)
  64.                             .build();
  65.                     EnergyMeasure production2 = EnergyMeasure.builder()
  66.                             .period(new InstantPeriod(period2From, period2To))
  67.                             .expiry(expiry)
  68.                             .production(1.6)
  69.                             .networkConsumption(0.0)
  70.                             .selfConsumption(1.6)
  71.                             .build();
  73.                     Map<String, Units> units = ImmutableMap.<String, Units>builder()
  74.                             .put("production", new Units(unitsValue))
  75.                             .put("selfConsumption", new Units(unitsValue))
  76.                             .put("networkConsumption", new Units(unitsValue))
  77.                             .build();
  79.                     EnergyProductionMessage message = EnergyProductionMessage.builder()
  80.                             .deviceId(deviceId)
  81.                             .username(username)
  82.                             .tenantId(tenantId)
  83.                             .units(units)
  84.                             .solarEnergies(Arrays.asList(production1, production2))
  85.                             .build();
  87.                     List<ProductionEntity> productions = productionMessageConverter.convert(message);
  88.                     productions.forEach(productionEntity -> productionEntity.setExpiry(computeTtl(productionEntity.getBeginTs())));
  89.                     result.addAll(productions);
  90.                     sendMessage(message);
  91.                 });
  92.         return result;
  93.     }
  95.     private long computeTtl(Instant fromBeginTs) {
  96.         return fromBeginTs.atZone(ZoneId.of("UTC")).plusMonths(25).toEpochSecond()
  97.                 - NOW.toEpochMilli() / 1000;
  98.     }
  100.     public void sendMessage(EnergyProductionMessage message) {
  101.         sink.send(getMessage(message));
  102.     }
  104.     public Message<EnergyProductionMessage> getMessage(EnergyProductionMessage message) {
  105.         return MessageBuilder.withPayload(message).build();
  106.     }
  108.     public void sendMessage(String payload) throws IOException {
  109.         EnergyProductionMessage energyProductionMessage = objectMapper.readValue(payload,
  110.                 EnergyProductionMessage.class);
  111.         Message<EnergyProductionMessage> message = MessageBuilder.withPayload(energyProductionMessage).build();
  112.         sink.send(message);
  113.     }
  114. }

型

ProductionMessageConsumerIT.java

  1. package com.enterprise.production.event.consumer;
  3. import com.enterprise.production.Application;
  4. import com.enterprise.production.client.ClientException;
  5. import com.enterprise.production.model.message.EnergyMeasure;
  6. import com.enterprise.production.model.message.EnergyProductionMessage;
  7. import com.enterprise.production.model.message.Units;
  8. import com.enterprise.production.model.time.InstantPeriod;
  9. import com.enterprise.production.repository.ProductionRepository;
  10. import com.enterprise.production.repository.entity.ProductionEntity;
  11. import com.google.common.collect.ImmutableMap;
  12. import org.apache.commons.lang3.builder.EqualsBuilder;
  13. import org.json.JSONArray;
  14. import org.json.JSONException;
  15. import org.json.JSONObject;
  16. import org.junit.jupiter.api.Assertions;
  17. import org.junit.jupiter.api.Test;
  18. import org.springframework.beans.factory.annotation.Autowired;
  19. import org.springframework.boot.test.autoconfigure.jdbc.AutoConfigureTestDatabase;
  20. import org.springframework.boot.test.autoconfigure.orm.jpa.AutoConfigureDataJpa;
  21. import org.springframework.boot.test.autoconfigure.orm.jpa.AutoConfigureTestEntityManager;
  22. import org.springframework.boot.test.context.SpringBootTest;
  23. import org.springframework.cloud.contract.wiremock.AutoConfigureWireMock;
  24. import org.springframework.context.annotation.Bean;
  25. import org.springframework.context.annotation.Configuration;
  26. import org.springframework.context.annotation.Primary;
  27. import org.springframework.core.io.Resource;
  28. import org.springframework.core.io.ResourceLoader;
  29. import org.springframework.test.context.ActiveProfiles;
  31. import java.io.IOException;
  32. import java.nio.file.Files;
  33. import java.time.Clock;
  34. import java.time.Instant;
  35. import java.time.ZoneId;
  36. import java.time.ZoneOffset;
  37. import java.time.temporal.ChronoUnit;
  38. import java.util.*;
  39. import java.util.concurrent.TimeUnit;
  40. import java.util.stream.Stream;
  42. import static com.github.tomakehurst.wiremock.client.WireMock.*;
  43. import static org.junit.jupiter.api.Assertions.assertThrows;
  45. @SpringBootTest(classes = {Application.class, ProductionMessageConsumerIT.TestConfig.class})
  46. @AutoConfigureWireMock(port = 0)
  47. @ActiveProfiles("it")
  48. public class ProductionMessageConsumerIT {
  50.     public static final Instant NOW = Instant.parse("2018-07-15T15:15:21Z");
  51.     public static final String CAS_TABLE_ENERGIES = "energies_v2";
  52.     public static final String CAS_TABLE_ENERGIES30m = "energies30m_v2";
  53.     @Autowired
  54.     ProductionRepository productionRepository;
  55.     @Autowired
  56.     StreamUtils streamUtils;
  57.     @Autowired
  58.     ResourceLoader resourceLoader;
  60. // Some Tests ...
  62.     @Configuration
  63.     static class TestConfig {
  64.         @Bean
  65.         @Primary
  66.         public Clock clockIT() {
  67.             return Clock.fixed(NOW, ZoneOffset.UTC); // timestamp 1531667721
  68.         }
  69.     }
  70. }

型

问题

接口Sink不在Spring-cloud-stream https://javadoc.io/doc/org.springframework.cloud/spring-cloud-stream/4.0.0/org/springframework/cloud/stream/messaging/package-summary.html的新版本4.0.0上，所以我需要将代码中过去的Sink替换为Spring-cloud-stream的新类DirectWithAttributesChannel，但是当我这样做时，我得到了这个错误：

  1. Field sink in com.enterprise.production.event.consumer.StreamUtils required a bean of type 'org.springframework.cloud.stream.messaging.DirectWithAttributesChannel' that could not be found.
  3. The injection point has the following annotations:
  4.     - @org.springframework.beans.factory.annotation.Autowired(required=true)
  6. Action:
  8. Consider defining a bean of type 'org.springframework.cloud.stream.messaging.DirectWithAttributesChannel' in your configuration.

型

提问

有人知道如何将Sink接口从spring-cloud-stream v2迁移到spring-cloud-stream v4吗？

spring

1条答案

按热度按时间
anauzrmj

anauzrmj1#

由EnableBinding,Sink,Source和Processor组件组成的基于注解的编程模型早已被弃用，并在Spring Cloud Stream的4.x版本中完全删除。您应该迁移到基于Spring Cloud Function架构的功能模型。在功能模型中，Sink可以转换为java.util.function.Consumer，一个Source作为一个java.util.function.Supplier和一个Processor通过java.util.function.Function。除了这些，还有一个StreamBridge API用于按需发送数据。所有这些都可以在参考文档中找到。由于您使用的是4.0.x，这里是最新4.0.x的参考文档：https://docs.spring.io/spring-cloud-stream/docs/4.0.3/reference/html/
在旧模型中，当您使用Sink时，您通常将其与EnableBinding(Sink.class)一起使用，然后使用类似StreamListener的东西来接收通过sink发送的消息。sink负责从消息传递中间件消费记录并发送到输入通道，然后由StreamListener侦听。所有这些基于注解的编程模型和组件都是从框架中删除，优先于功能模型。通过查看您提供的代码示例，看起来您正在尝试生成一些数据，然后将其发送到输入通道。如果您的用例到此结束，那么我建议直接使用Spring Integration，因为它对消息传递通道级别的工作提供了一流的支持。Spring Cloud Stream是一个更高的-在内部使用Spring Integration的抽象层。
如果您想在应用程序中生成数据后将其发送到中间件目的地，则可以使用StreamBridge。

  1. @Autowired 
  2. StreamBridge streamBridge;
  3. ...
  4. streamBridge.send("binding-name", data);
  5. }

字符串
或者如果你想简单地监听中间件目的地（例如Kafka主题），那么你可以使用标准的java.util.function.Consumer（或者如果你想把数据发送到另一个目的地，则使用Function）。

  1. @Bean
  2. public Consumer<Data> consume() {
  3.  data -> {
  4.    // custom code, maybe StreamBridge.send...
  5.  }
  6. }
  8. spring.cloud.stream.bindings.consume-in-0.destination: <Middleware destination>

型
或

  1. @Bean
  2. public Function<DataIn, DataOut> function() {
  3.  return in -> {
  4.   ...
  5.   return dataOut;
  6.  }
  7. }
  9. spring.cloud.stream.bindings.function-in-0.destination: <Middleware destination>
  11. spring.cloud.stream.bindings.function-out-0.destination: <Middleware destination>

型

展开查看全部
赞(0）分享  回复(0）举报  2024-01-05
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com