升级到springboot2、reactor 3.5、kafka binder 2.0.0版本和kafka client 1.0.1后,其中一个模块无法工作。我花了5天的时间来制作它,可能读了所有相关的主题,但找不到这种行为的原因。
主要类别:
@Slf4j
@EnableI18N
@EnableSideBar
@ComponentScan
@SpringBootConfiguration
@EnableConfigurationProperties
@EnableAspectJAutoProxy(proxyTargetClass = true)
public class View
{
public static void main(String[] args)
{
new SpringApplicationBuilder(View.class)
.profiles("production")
.bannerMode(Banner.Mode.OFF)
.headless(true)
.application()
.run(args);
log.info("\nhttp://localhost:8083/\n");
}
}
配置标记:
@Configuration
@Profile("production")
@EnableBinding(OffersChannel.class)
class ProductionOffersConfiguration
{
}
通道接口:
public interface OffersChannel
{
String OFFERS_OBTAIN = "offersObtain";
String OFFERS_OBTAIN_REQUEST= "offersObtainRequest";
@Input(OFFERS_OBTAIN)
SubscribableChannel offersChannel();
@Output(OFFERS_OBTAIN_REQUEST)
MessageChannel offersRequestChannel();
}
还有adminui类,更新依赖项之前最重要的是,当这个类被初始化时,我能够看到一个登录控制台,上面说我订阅了通道,现在什么都没有发生:
@Slf4j
@Push
@Theme("${view.default-theme}")
@SpringUI(path = WebsiteMapping.ADMIN)
@RequiredArgsConstructor
public class AdminUI extends UI
{
MessageChannel offersObtainRequest;
Grid<Ad> adGrid = createAdGrid();
private List<Ad> ads = new LinkedList<>();
ConnectionService connectionService;
@Override
@SneakyThrows
protected void init(VaadinRequest vaadinRequest)
{
setContent(splitPane());
adGrid.setItems(ads);
}
private VerticalSplitPanel splitPane()
{
VerticalSplitPanel verticalSplitPanel = new VerticalSplitPanel();
verticalSplitPanel.setSplitPosition(10, Unit.PERCENTAGE);
verticalSplitPanel.setFirstComponent(buttonsLayout());
verticalSplitPanel.setSecondComponent(adGrid);
return verticalSplitPanel;
}
private Layout buttonsLayout()
{
HorizontalLayout layout = new HorizontalLayout();
layout.setMargin(true);
layout.addComponent(requestMoreOffersButton());
ThemeSelectorComboBox themeSelectorComboBox = new ThemeSelectorComboBox();
layout.addComponent(themeSelectorComboBox);
layout.setComponentAlignment(themeSelectorComboBox, Alignment.MIDDLE_RIGHT);
return layout;
}
@PostConstruct
private void createGridProperties()
{
adGrid.setSizeFull();
adGrid.addColumn(Ad::getTitle).setCaption("Title");
adGrid.addColumn(Ad::getLocation).setCaption("Location");
adGrid.addColumn(Ad::getHref).setCaption("Href");
}
@StreamListener
public void fetchAdsFrom(@Input(OffersChannel.OFFERS_OBTAIN) Flux<Ad> fluxAd)
{
fluxAd.subscribe(this::displayOfferInGrid);
}
private void displayOfferInGrid(Ad ad)
{
ads.add(ad);
adGrid.setItems(ads);
}
private Button requestMoreOffersButton()
{
return new Button("Request 10 more offers", this::requestMoreOffers);
}
private Button startServiceButton(String caption, String url, String message)
{
return new Button(caption, buttonClickedEvent -> startService(url, message));
}
private void startService(String url, String message)
{
ConnectionRequest startProviderRequest = ConnectionRequest
.builder()
.url(url)
.build();
connectionService
.getForHtml(startProviderRequest)
.thenAccept(serviceStarted -> Notification.show(message));
}
private void requestMoreOffers(Event event)
{
offersObtainRequest.send(new GenericMessage(new AdBroadcastRequest(ads.size(),10)));
}
private Grid<Ad> createAdGrid()
{
return new Grid<>();
}
}
应用程序.yml:
spring:
cloud:
stream:
bindings:
offersObtainRequest:
destination: adsBroadcastRequestes
binder: kafka
group: adsBroadcastRequestsProducer
offersObtain:
destination: adsBroadcast
binder: kafka
group: adsConsumer
堆栈跟踪:
2018-04-07 11:27:12.010 DEBUG o.s.retry.support.RetryTemplate : Retry: count=0
2018-04-07 11:27:12.010 DEBUG o.s.integration.channel.DirectChannel : preSend on channel 'offersObtain', message: GenericMessage [payload=byte[227], headers={kafka_offset=12, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@11d6acc8, deliveryAttempt=1, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, contentType=application/json, kafka_receivedTopic=adsBroadcast, kafka_receivedTimestamp=1523093225984}]
2018-04-07 11:27:12.010 DEBUG o.s.r.backoff.ExponentialBackOffPolicy : Sleeping for 1000
2018-04-07 11:27:13.011 DEBUG o.s.retry.support.RetryTemplate : Checking for rethrow: count=1
2018-04-07 11:27:13.011 DEBUG o.s.retry.support.RetryTemplate : Retry: count=1
2018-04-07 11:27:13.011 DEBUG o.s.integration.channel.DirectChannel : preSend on channel 'offersObtain', message: GenericMessage [payload=byte[227], headers={kafka_offset=12, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@11d6acc8, deliveryAttempt=2, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, contentType=application/json, kafka_receivedTopic=adsBroadcast, kafka_receivedTimestamp=1523093225984}]
2018-04-07 11:27:13.011 DEBUG o.s.r.backoff.ExponentialBackOffPolicy : Sleeping for 2000
2018-04-07 11:27:13.909 DEBUG o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=adsConsumer] Sending Heartbeat request to coordinator Kacper-PC:9092 (id: 2147483647 rack: null)
2018-04-07 11:27:14.138 DEBUG o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=adsConsumer] Received successful Heartbeat response
2018-04-07 11:27:15.011 DEBUG o.s.retry.support.RetryTemplate : Checking for rethrow: count=2
2018-04-07 11:27:15.011 DEBUG o.s.retry.support.RetryTemplate : Retry: count=2
2018-04-07 11:27:15.011 DEBUG o.s.integration.channel.DirectChannel : preSend on channel 'offersObtain', message: GenericMessage [payload=byte[227], headers={kafka_offset=12, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@11d6acc8, deliveryAttempt=3, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, contentType=application/json, kafka_receivedTopic=adsBroadcast, kafka_receivedTimestamp=1523093225984}]
2018-04-07 11:27:15.011 DEBUG o.s.retry.support.RetryTemplate : Checking for rethrow: count=3
2018-04-07 11:27:15.011 DEBUG o.s.retry.support.RetryTemplate : Retry failed last attempt: count=3
2018-04-07 11:27:15.012 DEBUG o.s.i.h.a.ErrorMessageSendingRecoverer : Sending ErrorMessage: failedMessage: GenericMessage [payload=byte[227], headers={kafka_offset=12, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@11d6acc8, deliveryAttempt=3, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, contentType=application/json, kafka_receivedTopic=adsBroadcast, kafka_receivedTimestamp=1523093225984}]
org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application.offersObtain'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[227], headers={kafka_offset=12, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@11d6acc8, deliveryAttempt=3, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, contentType=application/json, kafka_receivedTopic=adsBroadcast, kafka_receivedTimestamp=1523093225984}]
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:463) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:407) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:181) ~[spring-messaging-5.0.4.RELEASE.jar:5.0.4.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:160) ~[spring-messaging-5.0.4.RELEASE.jar:5.0.4.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47) ~[spring-messaging-5.0.4.RELEASE.jar:5.0.4.RELEASE]
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:108) ~[spring-messaging-5.0.4.RELEASE.jar:5.0.4.RELEASE]
at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:203) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]
at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.access$300(KafkaMessageDrivenChannelAdapter.java:70) ~[spring-integration-kafka-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:387) ~[spring-integration-kafka-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:364) ~[spring-integration-kafka-3.0.3.RELEASE.jar:3.0.3.RELEASE]
at org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter.lambda$onMessage$0(RetryingMessageListenerAdapter.java:120) [spring-kafka-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287) [spring-retry-1.2.2.RELEASE.jar:na]
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:211) [spring-retry-1.2.2.RELEASE.jar:na]
at org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter.onMessage(RetryingMessageListenerAdapter.java:114) [spring-kafka-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter.onMessage(RetryingMessageListenerAdapter.java:40) [spring-kafka-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1001) [spring-kafka-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:981) [spring-kafka-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:932) [spring-kafka-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:801) [spring-kafka-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:689) [spring-kafka-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_151]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_151]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_151]
Caused by: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:138) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:105) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:73) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]
... 23 common frames omitted
3条答案
按热度按时间kyvafyod1#
嗯,里面发生了很多事。我是说所有非春日云流的东西。很难理解。无论如何,你的
@StreamListener
在任何spring管理的配置类中都没有定义,因此不会获取它。你可以把它移到ProductionOffersConfiguration
或者View
或者任何其他spring管理的配置类。另外,请考虑阅读本快速教程https://docs.spring.io/spring-cloud-stream/docs/elmhurst.release/reference/htmlsingle/#_quick_start (5分钟)更好地理解 Spring 云流的力学
hsvhsicv2#
你可以用
@StreamListener
注解,但注解内有目标参数。为我工作,类似于你的设置(而不是通量我收到一个json字符串)。
这可能是因为你必须为flux消息编写自己的转换器,我不确定。
3zwjbxry3#
在我的例子中,我声明了带有@input channel的spring云steam接口
但我没有使用任何@streamlistener来处理这个目标,所以我得到了这样的错误
解决方案:我删除了@input声明,只留下@output(“out”)一个。
仅供参考:我的输入配置在其他单独的服务。