java - 使用异步拉取从 Google PubSub 连续接收消息
问题描述
在此链接的帮助下,我设法创建了一个小型 Java 应用程序,它可以在一分钟内提取已发布的消息。我的实现看起来像这样。
public static void eventListener() throws InterruptedException {
MessageReceiver receiver = new MessageReceiver() {
@Override
public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
System.out.println("Received message: " + message.getData().toStringUtf8());
consumer.ack();
}
};
//Subscriber subscriber = null;
try {
subscriber = Subscriber.newBuilder(subscription, receiver)
.setCredentialsProvider(FixedCredentialsProvider.create(creds)).build();
subscriber.addListener(new Subscriber.Listener() {
@Override
public void failed(Subscriber.State from, Throwable failure) {
// Handle failure. This is called when the Subscriber encountered a fatal error
// and is
// shutting down.
System.err.println(failure);
}
}, MoreExecutors.directExecutor());
subscriber.startAsync().awaitRunning();
// In this example, we will pull messages for one minute (60,000ms) then stop.
// In a real application, this sleep-then-stop is not necessary.
// Simply call stopAsync().awaitTerminated() when the server is shutting down,
// etc.
Thread.sleep(60000);
} finally {
if (subscriber != null) {
subscriber.stopAsync().awaitTerminated();
}
}
}
当我调用这个方法时main
public static void main(String[] args) throws InterruptedException {
eventListener();
}
并将对象上传到我的谷歌云存储,程序打印发布者发送的消息,像这样
Received message: {
"kind": "storage#object",
"id": "roshanbucket/stones.jpg/1553765105996166",
"selfLink": "https://www.googleapis.com/storage/v1/b/roshanbucket/o/stones.jpg",
"name": "stones.jpg",
"bucket": "roshanbucket",
"generation": "1553765105996166",
"metageneration": "1",
"contentType": "image/jpeg",
"timeCreated": "2019-03-28T09:25:05.995Z",
"updated": "2019-03-28T09:25:05.995Z",
"storageClass": "STANDARD",
"timeStorageClassUpdated": "2019-03-28T09:25:05.995Z",
"size": "137256",
"md5Hash": "1GmpUnGeiW+/KU+0U8c8Wg==",
"mediaLink": "https://www.googleapis.com/download/storage/v1/b/roshanbucket/o/stones.jpg?generation=1553765105996166&alt=media",
"crc32c": "FMaEGg==",
"etag": "CIaj1InCpOECEAE="
}
自程序执行后的一分钟内,它会打印对象上传帐户上收到的所有消息,然后停止。要在一分钟后接收事件消息,我需要重新启动应用程序。现在,我想做的是连续运行监听器,所以,我尝试eventListener()
在 main 方法内的无限循环中运行该方法,就像这样
public static void main(String[] args) throws InterruptedException {
while(true) {
eventListener();
}
}
有了这个,我似乎能够在每次上传后立即收到事件消息,无论我何时上传对象。但是,每隔一段时间,它就会抛出这个堆栈跟踪。
Mar 28, 2019 12:56:34 PM io.grpc.internal.ManagedChannelOrphanWrapper$ManagedChannelReference cleanQueue
SEVERE: *~*~*~ Channel ManagedChannelImpl{logId=6, target=pubsub.googleapis.com:443} was not shutdown properly!!! ~*~*~*
Make sure to call shutdown()/shutdownNow() and wait until awaitTermination() returns true.
java.lang.RuntimeException: ManagedChannel allocation site
at io.grpc.internal.ManagedChannelOrphanWrapper$ManagedChannelReference.<init>(ManagedChannelOrphanWrapper.java:103)
at io.grpc.internal.ManagedChannelOrphanWrapper.<init>(ManagedChannelOrphanWrapper.java:53)
at io.grpc.internal.ManagedChannelOrphanWrapper.<init>(ManagedChannelOrphanWrapper.java:44)
at io.grpc.internal.AbstractManagedChannelImplBuilder.build(AbstractManagedChannelImplBuilder.java:440)
at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.createSingleChannel(InstantiatingGrpcChannelProvider.java:223)
at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.createChannel(InstantiatingGrpcChannelProvider.java:164)
at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.getTransportChannel(InstantiatingGrpcChannelProvider.java:156)
at com.google.api.gax.rpc.ClientContext.create(ClientContext.java:157)
at com.google.cloud.pubsub.v1.stub.GrpcSubscriberStub.create(GrpcSubscriberStub.java:260)
at com.google.cloud.pubsub.v1.Subscriber.doStart(Subscriber.java:268)
at com.google.api.core.AbstractApiService$InnerService.doStart(AbstractApiService.java:148)
at com.google.common.util.concurrent.AbstractService.startAsync(AbstractService.java:225)
at com.google.api.core.AbstractApiService.startAsync(AbstractApiService.java:120)
at com.google.cloud.pubsub.v1.Subscriber.startAsync(Subscriber.java:260)
at listener.AsynchronousPull.eventListener(AsynchronousPull.java:57)
at listener.AsynchronousPull.main(AsynchronousPull.java:74)
但是,它仍然会在每次上传后打印消息,同时不时抛出堆栈跟踪。我对 s 没有太多经验,thread
我非常感谢您在解决此问题方面提供的帮助。
解决方案
在这里调用eventListener()
一个紧密的循环不是你想要做的。这将创建许多订阅者的新实例,这些实例接收消息,每个实例都存在 60 秒。您想要的是让您创建的订阅者的单个实例一直存在到您想要关闭它的时间。通常,您可以通过创建订阅者并通过awaitTerminated()
.
上面的代码将被更改为:
public static void eventListener() throws InterruptedException {
MessageReceiver receiver = new MessageReceiver() {
@Override
public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
System.out.println("Received message: " + message.getData().toStringUtf8());
consumer.ack();
}
};
Subscriber subscriber = null;
try {
subscriber = Subscriber.newBuilder(subscription, receiver)
.setCredentialsProvider(FixedCredentialsProvider.create(creds)).build();
subscriber.addListener(new Subscriber.Listener() {
@Override
public void failed(Subscriber.State from, Throwable failure) {
// Handle failure. This is called when the Subscriber encountered a fatal error
// and is
// shutting down.
System.err.println(failure);
}
}, MoreExecutors.directExecutor());
subscriber.startAsync().awaitRunning();
subscriber.awaitTerminated();
} finally {
if (subscriber != null) {
subscriber.stopAsync().awaitTerminated();
}
}
}
public static void main(String[] args) throws InterruptedException {
eventListener();
}
如果您只想在应用程序终止时停止订阅者并且不想打扰任何额外的清理,那么上面的代码将起作用,允许订阅者运行并接收消息,直到发生错误或应用程序关闭。如果您想对应用程序的干净终止进行一些清理,例如,您想确保任何消息已经被receiveMessage
run 处理到完成,那么您可以附加一个关闭挂钩来捕获此类终止(尽管它不会全部运行情况)。在这个钩子中,你会调用stopAsync()
. 例如,您可以在try
块之前插入以下内容:
Runtime.getRuntime().addShutdownHook(new Thread() {
public void run() {
subscriber.stopAsync().awaitTerminated();
}
});
推荐阅读
- python - 如何在数据框中删除一个空行
- json - 从 json 文件映射时,peewee 模型的正确定义是什么?
- android - 刷新时设置微调器的当前项
- javascript - 筛选
- java - 如何从包含数字和字母的字符串中读取一个 int 数字?
- html - "GET /media/img/image.png HTTP/1.1" 404 2575
- r - 跨多个变量复制分组函数以生成许多新变量
- r - Rmarkdown 不会将蒙特塞拉特字体编织为 PDF
- android - startActivity 在 Android 11 中引发异常
- java - Antlr4 - 获取令牌名称