使用 Spring 集成实现消费者和生产者系统

Question

我想使用 Spring SubscribableChannel 实现单个生产者..多个消费者问题。我无法解决这个问题。

例如，我有一个用户作为生产者，他的朋友作为消费者。所以当生产者用户做某事时，它必须发布给他的朋友。

真正的问题是生产者和消费者的关系是动态的，因为我可以随时取消与用户的好友关系，这意味着它不应该再听我的活动了。所以我想要某种 java 配置，我可以在其中订阅和取消订阅生产者的消费者，还必须有一些东西可以将消费者与生产者分离，以防生产者不再存在......

我在设计频道时遇到了困难，因为会有number of channel equal to number of users. 我也不知道如何设计 Spring Integration 附带的 MessageHandler。

有没有一些指向这个方向的例子。

到目前为止我所做的是我有一个配置 bean

@Configuration
public class ChannelConfig {

    @Bean
    public SubscribableChannel subscribeChannel() {
        PublishSubscribeChannel channel = new PublishSubscribeChannel(getAsyncExecutor());
        return channel;
    }

    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(2);
        executor.setMaxPoolSize(50);
        executor.setQueueCapacity(10000);
        executor.setThreadNamePrefix("ChannelExecutor-");
        executor.initialize();
        return executor;
    }
}

我将注入，但除此之外，这将是唯一的唯一渠道。我希望每个用户都有一个频道。

Answer 1

为什么要为每个用户提供单独的频道？这就是发布/订阅频道的全部意义所在；只需subscribe() 为该频道的每个用户创建一个新的MessageHandler。然后发送到该通道的任何消息都将被分派给每个消费者。

编辑：（基于下面的评论）。

在这种情况下，不要将频道声明为@Bean，只需以编程方式为每个生产者实例化它（或将其范围设置为prototype 并获取一个新实例）并订阅他的“朋友”。

Answer 2

我正在做这样的事情

public class ChannelCreator {

    private Map<String , SubscribableChannel> channels;
    ThreadPoolTaskExecutor executor;

    public ChannelCreator() {
        channels = new ConcurrentHashMap<String, SubscribableChannel>();
        prepareAsyncExecutor();
    }

    public void createChannel(String login) {
        PublishSubscribeChannel channel = new PublishSubscribeChannel(executor);
        channels.put(login, channel);
    }

    public SubscribableChannel getChannel(String login){
        SubscribableChannel channel = channels.get(login);
        if(channel == null){
            channel = new PublishSubscribeChannel(executor);
            channels.put(login, channel);
        }
        return channel;
    }

    public void removeChannel(String login){
        channels.remove(login);
    }

    public Executor prepareAsyncExecutor() {
        executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(2);
        executor.setMaxPoolSize(50);
        executor.setQueueCapacity(10000);
        executor.setThreadNamePrefix("ChannelExecutor-");
        executor.initialize();
        return executor;
    }
}

和

@Configuration
public class ChannelConfig {

    @Bean
    public ChannelCreator channelCreator() {
        ChannelCreator creator = new ChannelCreator();
        return creator;
    }
}