Spring Cloud Streams - несколько динамических мест назначения для источников и приемников - PullRequest
Новая
       83

Spring Cloud Streams - несколько динамических мест назначения для источников и приемников

1 голос
/ 07 марта 2019

В моей системе был запрос на изменение, который в настоящее время прослушивает несколько каналов и также отправляет сообщения на несколько каналов, но теперь имена получателей будут в базе данных и будут меняться в любое время. Мне трудно поверить, что я первый, кто столкнулся с этим, но я вижу там ограниченную информацию.

Все, что я нашел, это эти 2 ...
Динамическое назначение приемника: https://github.com/spring-cloud-stream-app-starters/router/tree/master/spring-cloud-starter-stream-sink-router, но как это работает для активного прослушивания этих каналов, как это делает @StreamListener?

Динамические источники назначения: https://github.com/spring-cloud/spring-cloud-stream-samples/blob/master/source-samples/dynamic-destination-source/,, который делает это 

@Bean
    @ServiceActivator(inputChannel = "sourceChannel")
    public ExpressionEvaluatingRouter router() {
        ExpressionEvaluatingRouter router = new ExpressionEvaluatingRouter(new SpelExpressionParser().parseExpression("payload.id"));
        router.setDefaultOutputChannelName("default-output");
        router.setChannelResolver(resolver);
        return router;
    }

Но что это за "payload.id"? И где там указаны пункты назначения ??

1 Ответ

0 голосов
/ 04 июня 2019

Не стесняйтесь улучшать мой ответ, надеюсь, он поможет другим.

Теперь код (он работал в моем отладчике). Это пример, а не производство готово!

Это как отправить сообщение в динамический пункт назначения 

import org.springframework.messaging.MessageChannel;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;


@Service
@EnableBinding
public class MessageSenderService {

    @Autowired
    private BinderAwareChannelResolver resolver;

    @Transactional
    public void sendMessage(final String topicName, final String payload) {
        final MessageChannel messageChannel = resolver.resolveDestination(topicName);
        messageChannel.send(new GenericMessage<String>(payload));
    }
}

И конфигурация для Spring Cloud Stream. 

spring:
  cloud:
    stream:
      dynamicDestinations: output.topic.1,output.topic2,output.topic.3

я нашел здесь https://docs.spring.io/spring-cloud-stream/docs/Elmhurst.RELEASE/reference/htmlsingle/index.html#dynamicdestination Весной будет работать Cloud Stream версии 2+. Я использую 2.1.2 

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
    <version>2.1.2.RELEASE</version>
</dependency>

Как использовать сообщение из динамического назначения

https://stackoverflow.com/a/56148190/4587961

* * Конфигурация тысяча двадцать-один * * тысяча двадцать-дв 
spring:
  cloud:
    stream:
      default:
        consumer:
          concurrency: 2
          partitioned: true
      bindings:
        # inputs
        input:
          group: application_name_group
          destination: topic-1,topic-2
          content-type: application/json;charset=UTF-8

Java-потребитель. 

@Component
@EnableBinding(Sink.class)
public class CommonConsumer {

    private final static Logger logger = LoggerFactory.getLogger(CommonConsumer.class);

    @StreamListener(target = Sink.INPUT)
    public void consumeMessage(final Message<Object> message) {
        logger.info("Received a message: \nmessage:\n{}", message.getPayload());
        // Here I define logic which handles messages depending on message headers and topic.
        // In my case I have configuration which forwards these messages to webhooks, so I need to have mapping topic name -> webhook URI.
    }
}
...