java - 在 Java 中使用 Azure Cosmos DB 发布/订阅的示例
问题描述
我需要一个带有 Azure Cosmos DB 的发布/订阅事件消息系统。我使用 Azure Cosmos DB Java SDK v4。
我尝试使用基于此示例的 ChangeFeedProcessor https://github.com/Azure-Samples/azure-cosmos-java-sql-api-samples/blob/main/src/main/java/com/azure/cosmos/examples /changefeed/SampleChangeFeedProcessor.java但它不像预期的那样工作。
我的问题:
- 饲料收集/容器继续增长。在所有活动节点都收到事件后,如何删除事件?
- 事件的延迟似乎相对较大。大约一分钟。
- 只有一个节点接收事件。这对于负载平衡来说似乎很有趣,但这不是我的用例。
解决方案
在 Java SDK 4.12.0 版本中,以下代码对我有用。但它使用来自驱动程序的测试代码。它可以在未来改变。
private static final String CHANNEL = "events";
private CosmosContainer collection;
private boolean stopped;
void start( String clientID ) {
CosmosContainerProperties props = new CosmosContainerProperties( CHANNEL, "/type" );
// delete all events after 60 seconds. All nodes should receive it in the meantime.
props.setDefaultTimeToLiveInSeconds( 60 );
collection = getOrCreateContainer( props );
Thread thread = new Thread( () -> {
String[] continuation = new String[1];
try {
while( !stopped ) {
// sample code: https://github.com/Azure/azure-sdk-for-java/blob/master/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerChangeFeedTest.java
CosmosChangeFeedRequestOptions options = continuation[0] == null ? //
CosmosChangeFeedRequestOptions.createForProcessingFromNow( FeedRange.forFullRange() ) : // initial value
CosmosChangeFeedRequestOptions.createForProcessingFromContinuation( continuation[0] ); // continue value
Iterator<EventPOJO> it = collection //
.queryChangeFeed( options, EventPOJO.class ) //
.handle( ( response ) -> continuation[0] = response.getContinuationToken() ) //
.iterator();
while( it.hasNext() ) {
EventPOJO event = it.next();
if( event.client != clientID ) {
// filter the own events
onMessage( event );
}
}
// poll interval
Thread.sleep( 1000 );
}
} catch( Throwable th ) {
if( !stopped ) {
PersistenceLogger.LOGGER.error( th );
}
}
}, CHANNEL );
thread.setDaemon( true );
thread.start();
}
<T> void send( T event, String clientID ) {
EventPOJO evt = new EventPOJO();
evt.id = ...
evt.client = clientID;
evt.type = event.getClass().getName();
evt.message = ...
collection.createItem( evt );
}
推荐阅读
- sql-server - 如何在 MS SQL 中从年初获取值
- javascript - 使用画布,如何在图像中切割带有嵌入阴影的孔?
- c++ - 运算符 const char* () 和显式复制 ctor
- mysql - 检查 MySQL 中 JSON 数据中是否存在值
- xamarin - Xamarin IOS 或 Xamarin Android 项目能否引用 .NET 5 或 .NET Core 项目
- reactjs - 从函数体而不是 useEffect(fn,[]) 获取数据时有什么缺点或问题?
- gatsby - Gastby 主页组件仅在 Prod 中的所有页面上双重渲染(不在本地/开发中)
- python - 条件语句和多个函数
- android - 将本机 AAR 导入本机 andorid - 无法访问的类错误
- javascript - 当导航栏定位时:已修复如何解决与其他组件重叠的问题?