我正在尝试使用pubsub模拟器发送推送消息,我也在使用spring boot,这是我的配置:
依赖关系:
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-gcp-starter-pubsub</artifactId>
</dependency>
我的豆:
@Configuration
@AutoConfigureBefore(value= GcpPubSubAutoConfiguration.class)
@EnableConfigurationProperties(value= GcpPubSubProperties.class)
public class EmulatorPubSubConfiguration {
@Value("${spring.gcp.pubsub.projectid}")
private String projectId;
@Value("${spring.gcp.pubsub.subscriptorid}")
private String subscriptorId;
@Value("${spring.gcp.pubsub.topicid}")
private String topicId;
@Bean
public Publisher pubsubEmulator() throws IOException {
String hostport = System.getenv("PUBSUB_EMULATOR_HOST");
ManagedChannel channel = ManagedChannelBuilder.forTarget(hostport).usePlaintext().build();
try {
TransportChannelProvider channelProvider =
FixedTransportChannelProvider.create(GrpcTransportChannel.create(channel));
CredentialsProvider credentialsProvider = NoCredentialsProvider.create();
// Set the channel and credentials provider when creating a `TopicAdminClient`.
// Similarly for SubscriptionAdminClient
TopicAdminClient topicClient =
TopicAdminClient.create(
TopicAdminSettings.newBuilder()
.setTransportChannelProvider(channelProvider)
.setCredentialsProvider(credentialsProvider)
.build());
ProjectTopicName topicName = ProjectTopicName.of(projectId, topicId);
// Set the channel and credentials provider when creating a `Publisher`.
// Similarly for Subscriber
return Publisher.newBuilder(topicName)
.setChannelProvider(channelProvider)
.setCredentialsProvider(credentialsProvider)
.build();
} finally {
channel.shutdown();
}
}
}
当然,我已将PUBSUB_EMULATOR_HOST系统变量设置为localhost:8085,仿真器在哪里运行
我创建了一个测试用的休息控制器:
@Autowired
private Publisher pubsubPublisher;
@PostMapping("/send1")
public String publishMessage(@RequestParam("message") String message) throws InterruptedException, IOException {
Publisher pubsubPublisher = this.getPublisher();
ByteString data = ByteString.copyFromUtf8(message);
PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();
ApiFuture<String> future = pubsubPublisher.publish(pubsubMessage);
//pubsubPublisher.publishAllOutstanding();
try {
// Add an asynchronous callback to handle success / failure
ApiFutures.addCallback(future,
new ApiFutureCallback<String>() {
@Override
public void onFailure(Throwable throwable) {
if (throwable instanceof ApiException) {
ApiException apiException = ((ApiException) throwable);
// details on the API exception
System.out.println(apiException.getStatusCode().getCode());
System.out.println(apiException.isRetryable());
}
System.out.println("Error publishing message : " + message);
System.out.println("Error publishing error : " + throwable.getMessage());
System.out.println("Error publishing cause : " + throwable.getCause());
}
@Override
public void onSuccess(String messageId) {
// Once published, returns server-assigned message ids (unique within the topic)
System.out.println(messageId);
}
},
MoreExecutors.directExecutor());
}
finally {
if (pubsubPublisher != null) {
// When finished with the publisher, shutdown to free up resources.
pubsubPublisher.shutdown();
pubsubPublisher.awaitTermination(1, TimeUnit.MINUTES);
}
}
return "ok";
@PostMapping("/pushtest")
public String pushTest(@RequestBody CloudPubSubPushMessage request) {
System.out.println( "------> message received: " + decode(request.getMessage().getData()) );
return request.toString();
}
我已经在模拟器中创建了主题和订阅,我遵循了本教程:
https://cloud.google.com/pubsub/docs/emulator
我使用以下命令在模拟器中将端点“ pushtest”设置为获取推送消息:
python subscriber.py PUBSUB_PROJECT_ID create-push TOPIC_ID SUBSCRIPTION_ID PUSH_ENDPOINT
但是当我运行测试时,没有达到“ / pushtest”端点,并且出现此错误:
任务java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@265d5d05[未完成,任务= java.util.concurrent.Executors$RunnableAdapter@a8c8be3[包装的任务= com.google.common.util.concurrent.TrustedListenableFutureTask@1a53c57c[状态=待处理,信息= [任务= [运行中= [尚未开始],com.google.api.gax.rpc.AttemptCallable @ 3866e1d0]]]]]]从java.util.concurrent.ScheduledThreadPoolExecutor@3f34809a拒绝[已终止,池大小= 0,活动线程= 0,排队的任务= 0,已完成的任务= 1]
为了确保模拟器运行正常,我使用以下命令在python中运行测试:
python publisher.py PUBSUB_PROJECT_ID publish TOPIC_ID
而且我在“ pushtest”端点中正确获取消息。
我不知道为什么对我的冒犯感到抱歉。
感谢您的帮助。