如何在Spring Integration中使用ZooKeeper实现分布式锁轮询器

Question

Spring Integration支持ZooKeeper，如中所述https://docs.spring.io/spring-integration/reference/html/zookeeper.html然而，这份文件太含糊了。

它建议在下面添加bean，但没有详细说明当节点被授予领导权时如何启动/停止轮询器。

@Bean
public LeaderInitiatorFactoryBean leaderInitiator(CuratorFramework client) {
return new LeaderInitiatorFactoryBean()
.setClient(client)
.setPath("/siTest/")
.setRole("cluster");
}

我们有关于如何使用zookeeper确保以下轮询器在集群中任何时候只运行一次的例子吗？

@Component
public class EventsPoller {
public void pullEvents() {
//pull events should be run by only one node in the cluster at any time
}
}

Answer 1

当LeaderInitiator成为领导者并且其领导权被撤销时，它会发出OnGrantedEvent和OnRevokedEvent。

请参阅https://docs.spring.io/spring-integration/reference/html/messaging-endpoints-chapter.html#endpoint-角色和下一个https://docs.spring.io/spring-integration/reference/html/messaging-endpoints-chapter.html#leadership-事件处理获取有关这些事件处理的更多信息，以及它如何影响特定角色中的组件。

尽管我同意Zookeper章节一定有SmartLifecycleRoleController章节的链接。请随时就此事提出JIRA，欢迎投稿！

更新

这就是我在测试中所做的：

@RunWith(SpringRunner.class)
@DirtiesContext
public class LeaderInitiatorFactoryBeanTests extends ZookeeperTestSupport {
private static CuratorFramework client;
@Autowired
private PollableChannel stringsChannel;
@BeforeClass
public static void getClient() throws Exception {
client = createNewClient();
}
@AfterClass
public static void closeClient() {
if (client != null) {
client.close();
}
}
@Test
public void test() {
assertNotNull(this.stringsChannel.receive(10_000));
}

@Configuration
@EnableIntegration
public static class Config {
@Bean
public LeaderInitiatorFactoryBean leaderInitiator(CuratorFramework client) {
return new LeaderInitiatorFactoryBean()
.setClient(client)
.setPath("/siTest/")
.setRole("foo");
}
@Bean
public CuratorFramework client() {
return LeaderInitiatorFactoryBeanTests.client;
}
@Bean
@InboundChannelAdapter(channel = "stringsChannel", autoStartup = "false", poller = @Poller(fixedDelay = "100"))
@Role("foo")
public Supplier<String> inboundChannelAdapter() {
return () -> "foo";
}
@Bean
public PollableChannel stringsChannel() {
return new QueueChannel();
}
}
}

我在日志中有这样的东西：

2018-12-14 10:12:33,542 DEBUG [Curator-LeaderSelector-0] [org.springframework.integration.support.SmartLifecycleRoleController] - Starting [leaderInitiatorFactoryBeanTests.Config.inboundChannelAdapter.inboundChannelAdapter] in role foo
2018-12-14 10:12:33,578 DEBUG [Curator-LeaderSelector-0] [org.springframework.integration.support.SmartLifecycleRoleController] - Stopping [leaderInitiatorFactoryBeanTests.Config.inboundChannelAdapter.inboundChannelAdapter] in role foo