为Azure Cosmos运行多个ChangeFeedProcessor实例

Question

我试图运行两个ChangeFeedProcessor实例，它们都指向同一个集合，并在一个Cosmos帐户中使用同一个租约集合。我在这两个实例中都指定了唯一的主机名

我的意图是根据逻辑分区（根据Microsoft文档）将提要负载分布在各个实例之间

当我试图启动第二个实例时，我在控制台中得到以下异常。

com.azure.data.cosmos.internal.changeFeed.implement.exceptionClassifier.classifyClientExceptionClassifier.java:56)com.azure.data.cosmos.internal.partitionprocessorimpl.lambda$run$0(PartitionProcessorimpl.java:115)reactor.core.publisher.monorunnable.block(Monorunnable.java:66)com.azure.data.cosmos.internal.chang在java.util.concurrent.threadPoolExecutor.runworker（threadPoolExecutor.java:1149)在java.util.concurrent.threadPoolExecutor.javer.run（threadPoolExecutor.java:624)在java.lang.thread.run（threadPoolExecutor.javer.run（threadPoolExecutor.java:748)在com.azure.data.cosmos.internal.changeed.implementation.Exception（Ternal.changeFeed.implementation.PartitionProcessorImpl.lambda$run$0(PartitionProcessorImpl.java:115)在reactor.core.publisher.monorunnable.block(Monorunnable.java:66)在com.azure.data.cosmos.internal.changeFeed.implement.PartitionSupervisorImpl$1.run（PartitionSupervisorImpl$1.run（PartitionSupervisorImplow:89)在24）在java.lang.thread.run(thread.java:748)在com.azure.data.cosmos.internal.changeFeed.implement.exceptionClassifier.ClassifyClientException(exceptionClassifier.java:56)在com.azure.data.cosmos.internal.changeFeed.implementation.partitionProcessorImpl.lambda$run$0(PartitionProcessorImpl.java:115)...etc

我使用了下面的maven依赖项

<dependency>
    <groupId>com.microsoft.azure</groupId>
    <artifactId>azure-cosmos</artifactId>
    <version>3.0.0</version>
    <exclusions>
        <exclusion>
            <artifactId>slf4j-api</artifactId>
            <groupId>org.slf4j</groupId>
        </exclusion>
        <exclusion>
            <artifactId>slf4j-log4j12</artifactId>
            <groupId>org.slf4j</groupId>
        </exclusion>
        <exclusion>
            <artifactId>guava</artifactId>
            <groupId>com.google.guava</groupId>
        </exclusion>
    </exclusions>
</dependency>

代码段

1. 创建ChangeFeedProcessors列表（用于数据库中找到的所有容器）

        //FEED DATABASE
        CosmosDatabase feedDatabase = cosmosClient.getDatabase(cosmosDbName);

        //LEASE DATABASE
        CosmosDatabase leaseDatabase = cosmosClient.getDatabase(cosmosDbName + LEASES);

        //List of Containers in Feed Database
        List<CosmosContainerProperties> containerPropertiesList = null;
        try {
            Flux<FeedResponse<CosmosContainerProperties>> containers = feedDatabase.readAllContainers();
            List<FeedResponse<CosmosContainerProperties>> list = containers.toStream().collect(Collectors.toList());//Abhishek Optimize
            containerPropertiesList = list.get(0).results();
        }
        catch (Exception e) {
            System.out.println("Fail to query Containers");
            throw new ServiceException("Fail to query Containers");
        }

containerPropertiesList.parallelStream().forEach(cosmosContainerProperties -> {
                //FEED CONTAINER
                String containerName = cosmosContainerProperties.getString("id");
                CosmosContainer feedContainer = feedDatabase.getContainer(containerName);

                //LEASE CONTAINER
                String leaseContainerName = containerName + "-leases";
                CosmosContainer leaseContainer = leaseDatabase.getContainer(leaseContainerName);

                //Building ChangeFeedProcessor for current Container
                ChangeFeedProcessorOptions changeFeedProcessorOptions = new ChangeFeedProcessorOptions();
                changeFeedProcessorOptions.startTime(OffsetDateTime.now());

                ChangeFeedProcessor changeFeedProcessor = null;
                try {
                    ChangeFeedProcessor.BuilderDefinition builderDefinition = ChangeFeedProcessor.Builder()                           
                            .hostName("Host1")//used Host2 in the other Host
                            .feedContainer(feedContainer)
                            .leaseContainer(leaseContainer)
                            .options(changeFeedProcessorOptions)
                            .handleChanges(docs -> {
                                documentChangeHandler.processChanges(containerName, docs);
                            });
                    changeFeedProcessor = builderDefinition.build();
                }
                catch (Exception e) {
                    System.out.println("Fail to initialize ChangeFeedProcessor for " + containerName);
                }
                resultList.add(changeFeedProcessor);

                System.out.println("processed:  " + leaseContainerName);
            });

public void startChangeFeed() {
        if (null != changeFeedProcessors && !changeFeedProcessors.isEmpty()) {
            changeFeedProcessors.parallelStream().forEach(processor->processor.start().block());
        }
        else {
            System.out.println("changeFeedProcessors list is empty.. probably changeFeedProcessor has not been setup yet");
        }
    }

Answer 1

从评论来看，该问题与VPN/代理或阻止所需端口范围的东西有关。

直接模式，需要在VPN/代理/防火墙中开放和配置一定的端口范围：

如果无法进行配置，可以切换到网关/HTTP模式。