我尝试使用“集群中的经典分布式发布订阅”的概念从微服务A中的senderActor发送消息我尝试将消息发布到content-topic，但我没有在位于微服务B中的ReceiverActor中接收到它。您是否了解可能导致此问题的原因？
谢谢你，谢谢！
A中的Conf-file
akka {
actor {
provider = "cluster"
}
remote {
artery {
enabled = on
canonical {
hostname = "localhost"
port = 2551
}
}
}
cluster {
seed-nodes = [
"akka://cluster-system@localhost:2551"
]
}
}SenderActor：
public class SenderActor extends AbstractActor {
ActorRef mediator = DistributedPubSub.get(getContext().system()).mediator();
@Override
public Receive createReceive() {
return receiveBuilder()
.match(
String.class,
in -> {
String out = in.toUpperCase();
mediator.tell(new DistributedPubSubMediator.Publish("content", out), getSelf());
})
.build();
}
}在微服务A中：
public static void main(String[] args) {
SpringApplication.run(MicroserviceAaApplication.class, args);
Config config = ConfigFactory.load("application.conf");
ActorSystem system = ActorSystem.create("cluster-system", config);
Cluster cluster = Cluster.get(system);
ActorRef publisher = system.actorOf(Props.create(SenderActor.class), "publisher");
publisher.tell("hello from microservice A ", null);
}B中Conf-file
akka {
actor {
provider = "cluster"
}
remote {
artery {
enabled = on
canonical {
hostname = "localhost"
port = 2552
}
}
}
cluster {
seed-nodes = [
"akka://cluster-system@localhost:2551"
]
}
}ReceiverActor类：
public class ReceiverActor extends AbstractActor {
LoggingAdapter log = Logging.getLogger(getContext().system(), this);
public ReceiverActor() {
ActorRef mediator = DistributedPubSub.get(getContext().system()).mediator();
mediator.tell(new DistributedPubSubMediator.Subscribe("content", getSelf()), getSelf());
}
@Override
public Receive createReceive() {
return receiveBuilder()
.match(String.class, msg -> {
log.info("Received message in receiver-actor-b: {}", msg);
})
.match(DistributedPubSubMediator.SubscribeAck.class,
msg -> log.info("SubscribeAck: {}", msg))
.build();
}
}微服务B中的主服务：
public static void main(String[] args) {
SpringApplication.run(MicroserviceBbApplication.class, args);
Config config = ConfigFactory.load("application.conf");
ActorSystem system = ActorSystem.create("cluster-system", config);
Cluster cluster = Cluster.get(system);
ActorRef subscriber = system.actorOf(Props.create(ReceiverActor.class), "subscriber");
}我尝试获取消息“来自微服务A的hello“，但我只得到B中的流日志：
2023-08-20T15:53:14.067+02:00 INFO 7788 --- [ main] c.e.M.MicroserviceBbApplication : Started MicroserviceBbApplication in 0.909 seconds (process running for 1.723)
2023-08-20T15:53:14.552+02:00 INFO 7788 --- [lt-dispatcher-5] akka.event.slf4j.Slf4jLogger : Slf4jLogger started
2023-08-20T15:53:14.900+02:00 INFO 7788 --- [lt-dispatcher-5] akka.remote.artery.ArteryTransport : Remoting started with transport [Artery tcp]; listening on address [akka://cluster-system@localhost:2552] with UID [6730186213375081245]
2023-08-20T15:53:14.925+02:00 INFO 7788 --- [lt-dispatcher-5] akka.cluster.Cluster : Cluster Node [akka://cluster-system@localhost:2552] - Starting up, Akka version [2.8.3] ...
2023-08-20T15:53:14.980+02:00 INFO 7788 --- [lt-dispatcher-5] akka.cluster.Cluster : Cluster Node [akka://cluster-system@localhost:2552] - Registered cluster JMX MBean [akka:type=Cluster]
2023-08-20T15:53:14.981+02:00 INFO 7788 --- [lt-dispatcher-5] akka.cluster.Cluster : Cluster Node [akka://cluster-system@localhost:2552] - Started up successfully
2023-08-20T15:53:15.006+02:00 INFO 7788 --- [t-dispatcher-13] akka.cluster.Cluster : Cluster Node [akka://cluster-system@localhost:2552] - No downing-provider-class configured, manual cluster downing required, see https://doc.akka.io/docs/akka/current/typed/cluster.html#downing
2023-08-20T15:53:15.624+02:00 INFO 7788 --- [t-dispatcher-13] akka.cluster.Cluster : Cluster Node [akka://cluster-system@localhost:2552] - Received InitJoinAck message from [Actor[akka://cluster-system@localhost:2551/system/cluster/core/daemon#1745370250]] to [akka://cluster-system@localhost:2552]
2023-08-20T15:53:15.675+02:00 INFO 7788 --- [t-dispatcher-13] akka.cluster.Cluster : Cluster Node [akka://cluster-system@localhost:2552] - Welcome from [akka://cluster-system@localhost:2551]
2023-08-20T15:53:16.077+02:00 INFO 7788 --- [lt-dispatcher-5] c.example.MicroserviceBB.ReceiverActor : SubscribeAck: SubscribeAck(Subscribe(content,None,Actor[akka://cluster-system/user/subscriber1#-339648796]))
2023-08-20T16:24:19.549+02:00 INFO 7788 --- [t-dispatcher-40] akka.actor.CoordinatedShutdown : Running CoordinatedShutdown with reason [JvmExitReason]
Process finished with exit code 130
1条答案
按热度按时间ulmd4ohb1#
重要的是要记住，分布式PubSub只会将消息传递给在中介器接收到发布消息时碰巧订阅的订阅者（我简化了一点，但这是直觉）。
在微服务A的
main中，您开始集群形成（形成集群的过程发生在后台），然后立即生成SenderActor并向其发送消息，该消息被转发到中介器。还要注意的是，集群可以只由一个节点组成（因为微服务A将自己定义为种子节点），所以在微服务B中的ReceiverActor订阅之前，您将消息发送到中介器;结果，该消息被丢弃。在这种情况下，在微服务A中有一个订阅不同pubsub主题的参与者可能会很有用;
ReceiverActor发布到它订阅了"content"主题的主题，这会导致微服务A中的订阅参与者tellSenderActor，它可以向该主题发布消息。顺便说一下，目前还不清楚微服务A和B是同一个微服务的示例，还是真正不同的微服务。如果是后者，则不建议使用Akka Cluster进行微服务之间的通信。