基于Java语言的springboot-ddd-framework设计模式通用组件框架源码

package com.zero.ddd.akka.event.publisher2.actor.broker; import java.time.Duration; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; import org.reactivestreams.Publisher; import com.google.common.cache.LoadingCache; import com.zero.ddd.akka.cluster.core.initializer.config.BlockingIODispatcherSelector; import com.zero.ddd.akka.cluster.core.initializer.serializer.SelfProtoBufObject; import com.zero.ddd.akka.event.publisher2.actor.ServiceKeyHolder; import com.zero.ddd.akka.event.publisher2.actor.consumer.EventSynchConsuemr.BatchPartitionEventCommand; import com.zero.ddd.akka.event.publisher2.actor.consumer.EventSynchConsuemr.EventSynchConsuemrEvent; import com.zero.ddd.akka.event.publisher2.actor.consumer.EventSynchConsuemr.PartitionEventCommand; import com.zero.ddd.akka.event.publisher2.domain.partitionEvent.PartitionEvent; import com.zero.ddd.akka.event.publisher2.domain.partitionEvent.PartitionEventStore; import com.zero.ddd.akka.event.publisher2.domain.synchronizerState.PartitionAssignState; import com.zero.ddd.akka.event.publisher2.domain.synchronizerState.SynchronizerState; import com.zero.ddd.akka.event.publisher2.event.EventSynchronizer; import com.zero.ddd.akka.event.publisher2.event.IRecordLastOffsetId; import com.zero.ddd.akka.event.publisher2.publisher.EventPublisherFactory; import com.zero.ddd.akka.event.publisher2.publisher.EventPublisherFactory.PartitionStoredEventWrapper; import com.zero.ddd.akka.event.publisher2.publisher.EventPublisherFactory.StoredEventWrapper; import com.zero.ddd.core.event.store.StoredEvent; import com.zero.helper.GU; import com.zero.helper.SimpleCacheBuilder; import akka.NotUsed; import akka.actor.typed.ActorRef; import akka.actor.typed.Behavior; import akka.actor.typed.PostStop; import akka.actor.typed.PreRestart; import akka.actor.typed.SupervisorStrategy; import akka.actor.typed.javadsl.ActorContext; import akka.actor.typed.javadsl.Behaviors; import akka.actor.typed.receptionist.Receptionist; import akka.actor.typed.receptionist.Receptionist.Listing; import akka.actor.typed.receptionist.ServiceKey; import akka.stream.KillSwitches; import akka.stream.Materializer; import akka.stream.UniqueKillSwitch; import akka.stream.javadsl.Flow; import akka.stream.javadsl.Keep; import akka.stream.javadsl.RetryFlow; import akka.stream.javadsl.Sink; import akka.stream.javadsl.Source; import akka.stream.typed.javadsl.ActorSink; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; import lombok.RequiredArgsConstructor; import lombok.ToString; import lombok.var; import lombok.extern.slf4j.Slf4j; /** * * @say little Boy, don't be sad. * @name Rezar * @time 2023-06-12 03:58:36 * @Desc 些年若许,不负芳华. * */ @Slf4j(topic = "event") public class EventSynchronizerPublishBroker { public static Behavior<EventSynchronizerBrokerEvent> create( String eventSynchronizerId, Materializer materizizer, IRecordLastOffsetId iRecordLastOffsetId, PartitionEventStore partitionEventStore, EventPublisherFactory eventPublisherFactory) { return Behaviors.setup( context -> new EventSynchronizerPublishBroker( eventSynchronizerId, materizizer, iRecordLastOffsetId, partitionEventStore, eventPublisherFactory, context) .startEventPublish()); } private final String eventSynchronizerId; private final PartitionEventPublisher partitionEventPublisher; private final ActorContext<EventSynchronizerBrokerEvent> context; private final ServiceKey<EventSynchConsuemrEvent> eventConsumerServiceKey; private final Map<String, ActorRef<EventSynchConsuemrEvent>> onlineWorker = new HashMap<>(); private SynchronizerState state; private LoadingCache<Integer, ActorRef<EventSynchronizerBrokerEvent>> sinkActorRouteActorCache; public EventSynchronizerPublishBroker( String eventSynchronizerId, Materializer materizizer, IRecordLastOffsetId iRecordLastOffsetId, PartitionEventStore partitionEventStore, EventPublisherFactory eventPublisherFactory, ActorContext<EventSynchronizerBrokerEvent> context) { this.context = context; this.eventSynchronizerId = eventSynchronizerId; this.partitionEventPublisher = new PartitionEventPublisher( materizizer, iRecordLastOffsetId, partitionEventStore, eventPublisherFactory); this.state = new SynchronizerState(); this.eventConsumerServiceKey = ServiceKeyHolder.eventConsumerServiceKey( this.eventSynchronizerId); this.initPartitionSinkActorCache(); this.registeAsSchedulerService(); this.subscribeEventConsumerService(); log.info( "事件主题:[{}] 启动成功!!", this.eventSynchronizerId); } private Behavior<EventSynchronizerBrokerEvent> startEventPublish() { return Behaviors .receive(EventSynchronizerBrokerEvent.class) .onMessage(EventSynchronizerInfoReport.class, this::onEventSynchronizerInfoReport) .onMessage(EventConsumerListingResponse.class, this::updateOnlineEventConsumerList) .onMessage(Passivate.class, this::onPassivate) .build(); } private void initPartitionSinkActorCache() { this.sinkActorRouteActorCache = SimpleCacheBuilder.instance( this::initPartitionSinkActor); } private ActorRef<EventSynchronizerBrokerEvent> initPartitionSinkActor( int partitionId) { return context.spawn( Behaviors.supervise( Behaviors.receive( EventSynchronizerBrokerEvent.class) .onMessage(BrokerRoutePartitionEvent.class, this::onBrokerRoutePartitionEvent) .onMessage(BrokerRouteBatchPartitionEvent.class, this::onBrokerRouteBatchPartitionEvent) .onMessage(BrokerRouteConsumerAckEvent.class, this::onBrokerRouteConsumerAckEvent) .onMessage(BrokerRoutePartitionSinkFail.class, this::onBrokerRoutePartitionSinkFail) .onMessage(BrokerRouteConsumerNeedCompletePartitionSync.class, this::onBrokerRouteConsumerNeedCompletePartitionSync) .onMessage(BrokerRouteConsumerNeedStartPartitionEventSync.class, this::onBrokerRouteConsumerNeedStartPartitionEventSync) .onSignal( PreRestart.class, signal -> { log.info( "事件主题:[" + this.eventSynchronizerId + "] 分区:[" + partitionId + "] 事件背压流启动!!!"); return Behaviors.same(); }) .onSignal( PostStop.class, signal -> { log.warn( "事件主题:[" + this.eventSynchronizerId + "] 分区:[" + partitionId + "] 事件背压流停止!!!"); return Behaviors.same(); }) .build()) .onFailure( Exception.class, SupervisorStrategy.restart()), // sink Actor名称 eventSynchronizerId + "-SinkActor-" + partitionId, BlockingIODispatcherSelector.defaultDispatcher()); } private ActorRef<EventSynchronizerBrokerEvent> partitionSinkActor( int partition) { return this.sinkActorRouteActorCache.getUnchecked(partition); } private Behavior<EventSynchronizerBrokerEvent> onPassivate( Passivate passivate) { this.partitionEventPublisher.shutdown(); return Behaviors.stopped(); } private Behavior<EventSynchronizerBrokerEvent> onBrokerRouteConsumerNeedStartPartitionEventSync( BrokerRouteConsumerNeedStartPartitionEventSync event) { event.replyOk(); return Behaviors.same(); } private Behavior<EventSynchronizerBrokerEvent> onBrokerRouteConsumerAckEvent( BrokerRouteConsumerAckEvent event) { this.partitionEventPublisher.storePartitionEventConsumedOffset( event.getPartition(), event.getConsumedOffset()); event.replyTo.tell(ACK.INSTANCE); return Behaviors.same(); } private Behavior<EventSynchronizerBrokerEvent> onBrokerRoutePartitionSin