如何使用 Zookeeper 和 Spring 集成负载平衡领导者答案

【问题标题】：How to load balance leader using zookeeper and spring integration如何使用 Zookeeper 和 Spring 集成负载平衡领导者
【发布时间】：2019-05-22 10:22:03
【问题描述】：

使用 spring 集成和 zookeeper，可以实现一个领导者来执行诸如轮询之类的活动。

但是我们如何将leader职责分配给集群中的所有节点来进行负载均衡呢？

鉴于下面的代码，一旦应用程序启动，我看到同一个节点正在维护领导角色并获取事件。我想将此活动分发到集群中的每个节点以更好地平衡负载。

有什么方法可以调度集群中的每个节点以循环方式获得领导权和撤销？

    @Bean
    public LeaderInitiatorFactoryBean fooLeaderInitiator(CuratorFramework client) {
        new LeaderInitiatorFactoryBean()
                .setClient(client)
                .setPath("/foofeed")
                .setRole("foo");
    }

    @Bean
    @InboundChannelAdapter(channel = "fooIncomingEvents", autoStartup = "false", poller = @Poller(fixedDelay = "5000"))
    @Role("foo")
    public FooTriggerMessageSource fooInboundChannelAdapter() {
        new FooMessageSource("foo")
    }

【问题讨论】：

标签： spring-integration apache-zookeeper spring-integration-dsl spring-cloud-zookeeper

【解决方案1】：

我可以使用以下代码模拟负载平衡。不确定这是否是正确的方法。我只能从集群中的一个节点一次看到 fetching events 日志语句。此代码在获得领导权并履行其职责后产生领导力。

@Bean
public LeaderInitiator fooLeaderInitiator(CuratorFramework client, 
        FooPollingCandidate fooPollingCandidate) {
    LeaderInitiator leader = new LeaderInitiator(client, fooPollingCandidate, zooKeeperNamespace)
    leader.start()
    leader
}

@Component
class FooPollingCandidate extends DefaultCandidate {
    final Logger log = LoggerFactory.getLogger(this.getClass());

    FooPollingCandidate() {
        super("fooPoller", "foo")
    }

    @Override
    void onGranted(Context ctx) {
        log.debug("Leadership granted {}", ctx)
        pullEvents()
        ctx.yield();
    }

    @Override
    void onRevoked(Context ctx) {
        log.debug("Leadership revoked")
    }

    @Override
    void yieldLeadership() {
        log.debug("yielding Leadership")
    }

    //pull events and drop them on any channel needed
    void pullEvents() {
        log.debug("fetching events")
        //simulate delay
        sleep(5000)
    }
}

【讨论】：

【解决方案2】：

您的建议是滥用领导选举技术，该技术旨在在当前领导失败时进行热故障转移，在每个事件后手动让出领导是一种反模式

您可能想要的是所有轮询器都处于活动状态的竞争轮询器，但使用共享存储来防止重复处理。

例如，如果您正在轮询共享目录以查找要处理的文件，则可以使用 FileSystemPersistentFileListFilter 和共享 MetadataStore（例如 zookeeper 实现）来防止多个实例处理同一个文件。

您可以对任何轮询消息源使用相同的技术（共享元数据存储）。

【讨论】：