java - 为集群环境创建列表

Question

在我的应用程序中有一个列表 publisherPostListenerList，它从 RabbitMQ 队列接收实时用户帖子，发送给订阅者/消费者。该列表是 ApplicationListener 类的一个属性，它监听 pubsub 队列的事件。下面的 Controller 方法通过 getter 方法获取列表元素，并根据逻辑将帖子推送给订阅者。

流程如下

用户写一个帖子 -> 帖子进入 DB + 队列 -> 来自队列的消息被添加到一个列表中，该列表是 publisherPostListenerList，将被推送给用户的订阅者。

正如我们所见，publisherPostListenerList 是 n 个并发请求的通用列表，因为 ApplicationListener 是单例。对于单个实例，设置工作正常，但在集群环境中会失败，因为每个节点都有自己的 publisherPostListenerList 列表。

我该如何处理这种情况？我不能使 ApplicationListener 类无状态 我需要列表来存储从队列中接收到的 post 元素。我是否将列表放在分布式内存缓存中？还是有其他常规方式？

ApplicationListener.java

@Component
public class ApplicationEventListener {

    private List<Post> publisherPostListenerList = new CopyOnWriteArrayList<Post>();

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

    @EventListener
    public void postSubmissionEventHandler(PostSubmissionEvent event) throws IOException {
        Post post = event.getPost();
        logger.debug("application published user post received " + post);
        publisherPostListenerList.add(post);
    }

    public List<Post> getPublisherPostListenerList() {
        return publisherPostListenerList;
    }

    public void setPublisherPostListenerList(List<Post> publisherPostListenerList) {
        this.publisherPostListenerList = publisherPostListenerList;
    }
}

向订阅者推送消息的 Controller 方法

@RequestMapping(value="/getRealTimeServerPushUserPosts")
    public SseEmitter getRealTimeServerPushUserPosts(@RequestParam("userId") int userId){
        SseEmitter sseEmitter = new SseEmitter();
        CustomUserDetail myUserDetails = currentUserAccessor.getCurrentLoggedInUser();
        User loggedInUser=myUserDetails.getUser();

        List<Integer> userPublisherIDList = this.userService.loadUserPublisherIdListWhichLoggedInUserFollows(loggedInUser);
        List<Post> postList =eventListener.getPublisherPostListenerList();


        for(Integer userPublisherId : userPublisherIDList){
            for(Post post:postList){
                    if((userPublisherId.intValue()) == (post.getUser().getUserId().intValue())){
                        try {
                        sseEmitter.send(post);
                        postList.remove(post); //removes the post for all the subscribers as the list acts as a global list.
                    } catch (IOException e) {
                        logger.error(e);
                    }
                }
             }
         }
        return sseEmitter;
    }

Answer 1

您可以使用 Hazelcast IList。它遵循 j.u.List 语义，适用于分布式/集群环境。

您可以找到文档 here和示例 here . 另一种选择是使用分布式 map 又名 IMap .

如果您对实现细节有任何具体问题，请告诉我。

谢谢