Actor模式理解与使用

一时失言乱红尘 2021-11-04 15:02 793阅读 0赞

>      最近学习ThingsBoard，其中大量使用了Actor设计模式，再这里做个Actor模式理解与使用的笔记
> 
>      Actor模式是一种并发模型，与另一种模型共享内存完全相反，Actor模型share nothing。所有的线程(或进程)通过消息传递的方式进行合作，这些线程(或进程)称为Actor。
> 
>       共享内存更适合单机多核的并发编程，而且共享带来的问题很多，编程也困难。随着多核时代和分布式系统的到来，共享模型已经不太适合并发编程，因此几十年前就已经出现的Actor模型又重新受到了人们的重视。

# **多线程编程-共享内存** #

--------------------

到了多核时代，有多个工人，这些工人共同使用一个仓库和车间，干什么都要排队。比如我要从一块钢料切出一块来用，我得等别人先用完。有个扳手，另一个人在用，我得等他用完。两个人都要用一个切割机从一块钢材切一块钢铁下来用，但是一个人拿到了钢材，一个人拿到了切割机，他们互相都不退让，结果谁都干不了活。

[![wKiom1WOKpHgGDoXAAFY8MuvOgA842.jpg][]][wKiom1WOKpHgGDoXAAFY8MuvOgA842.jpg 1]

# **多线程/分布式编程-Actor模型** #

--------------------

到了分布式系统时代，工厂已经用流水线了，每个人都有明确分工，这就是Actor模式。每个线程都是一个Actor，这些Actor不共享任何内存，所有的数据都是通过消息传递的方式进行的。

[![wKiom1WOKyPBDRjpAAGSeStUphQ651.jpg][]][wKiom1WOKyPBDRjpAAGSeStUphQ651.jpg 1]

如果用Actor模型实现统计素数个数，那么我们需要1个actor做原料的分发，就是提供要处理的整数，然后10个actor加工，每次从分发actor那里拿一个整数进行加工，最终把加工出来的半成品发给组装actor，组装actor把10个加工actor的结果汇总输出。

#  Actor模式的Springboot实现 #

了解过Akka或者Actor的人应该知道，这的确是一个很不错的框架，按照Akka官网的描述——使用Akka使得构建强有力的并发与分布式应用将更加容易。

与springboot集成，是因为我们想把ActorSystem、Actor等组件的创建纳入SpringBoot容器中，方便管理。大家都知道，ActorSystem的创建不是依赖new方式，而是通过create方法，所以我们需要写一个Bean来生产ActorSystem。另外Actor，它也是通过actorOf()方法创建的，所以我们也需要写生产Actor引用的方法，Akka提供了IndirectActorProducer接口，通过实现该接口，我们就可以实现DI（依赖注入）。集成springboot之后，ActorSystem范围内的依赖都会交给SpringBoot来管理，并且每个ActorSystem都会持有一个ApplicationContext。

下面我们开始来看看集成是怎样实现的：

在实现之前，我们需要加入相关依赖，如下：

<?xml version="1.0" encoding="UTF-8"?>
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 	<modelVersion>4.0.0</modelVersion>
 	<parent>
 		<groupId>org.springframework.boot</groupId>
 		<artifactId>spring-boot-starter-parent</artifactId>
 		<version>2.1.3.RELEASE</version>
 		<relativePath/> 
 	</parent>
 	<groupId>io.kunalk.spring</groupId>
 	<artifactId>springakka</artifactId>
 	<version>0.0.1-SNAPSHOT</version>
 	<name>springakka</name>
 	<description>Demo project for Spring Boot</description>
 
 	<properties>
 		<java.version>1.8</java.version>
 	</properties>
 
 	<dependencies>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-actuator</artifactId>
 		</dependency>
 
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>
 			<scope>test</scope>
 		</dependency>
 
 		<dependency>
 			<groupId>com.typesafe.akka</groupId>
 			<artifactId>akka-actor_2.11</artifactId>
 			<version>2.4.3</version>
 		</dependency>
 
 		<dependency>
 			<groupId>com.typesafe.akka</groupId>
 			<artifactId>akka-slf4j_2.11</artifactId>
 			<version>2.4.3</version>
 		</dependency>
 	</dependencies>
 
 	<build>
 		<plugins>
 			<plugin>
 				<groupId>org.springframework.boot</groupId>
 				<artifactId>spring-boot-maven-plugin</artifactId>
 			</plugin>
 		</plugins>
 	</build>
 
 </project>

第一步，实现IndirectActorProducer，用于生产Actor，既然是交给Spring管理，肯定少不了ApplicationContext对象和bean名称：

/**
 * 实现IndirectActorProducer，用于生产Actor
 *
 * 
 * Date 2019年8月8日 下午6:44:05
 * 
 * @author ieflex
 */
 public class SpringActorProducer implements IndirectActorProducer {
 
 	final private ApplicationContext applicationContext;
 	final private String actorBeanName;
 
 	public SpringActorProducer(ApplicationContext applicationContext, String actorBeanName) {
 		this.applicationContext = applicationContext;
 		this.actorBeanName = actorBeanName;
 	}
 
 	@Override
 	public Actor produce() {
 		return (Actor) applicationContext.getBean(actorBeanName);
 	}
 
 	@Override
 	public Class<? extends Actor> actorClass() {
 		return (Class<? extends Actor>) applicationContext.getType(actorBeanName);
 	}
 }

第二步，实现了DI之后，我们就需要构造Props对象，用来创建ActorRef：

/**
 * 扩展组件，ApplicationContext会在SpringBoot初始化的时候加载进来 
 * 构造Props,用于生产ActorRef
 */
 @Component
 public class SpringExtension implements Extension {
 
 	private ApplicationContext applicationContext;
 
 	public void initialize(ApplicationContext applicationContext) {
 		this.applicationContext = applicationContext;
 	}
 
 	/**
 	 * 该方法用来创建Props对象，依赖前面创建的SpringActorProducer,DI组件，获取到Props对象，我们就可以创建Actor bean对象
 	 *
 	 * @param beanName actor bean 名称
 	 * @return props
 	 */
 	public Props props(String actorBeanName) {
 		return Props.create(SpringActorProducer.class, applicationContext, actorBeanName);
 	}
 }

第三步，通过SpringExtProvider我们可以获取到SpringExt，通过SpringExt我们可以使用Props创建ActorRef对象，那么现在我们怎么来初始化ActorSystem，并扫描到纳入到容器的Actor呢？我们可以通过@Configuration来创建一个配置类：

/**
 * 创建ActorSystem，并将其放入到spring管理，初始化ApplicationContext
 *
 * 
 * Date 2019年8月8日 下午6:50:20
 * 
 * 
 * @author ieflex
 */
 @Configuration
 class ApplicationConfiguration {
 
 	@Autowired
 	private ApplicationContext applicationContext;
 
 	@Autowired
 	private SpringExtension springExtension;
 
 	@Bean
 	public ActorSystem actorSystem() {
 		ActorSystem actorSystem = ActorSystem.create("actor-system", akkaConfiguration());
 		springExtension.initialize(applicationContext);
 		return actorSystem;
 	}
 
 	@Bean
 	public Config akkaConfiguration() {
 		return ConfigFactory.load();
 	}
 }

下面，我们就来创建一个Actor示例，来验证集成是否成功，如下：

/**
     * 一般情况下我们的Actor都需要继承自UntypedActor，并实现其onReceive方法。onReceive用于接收消息，你可以在其中实现对消息的匹配并做不同的处理。
     * 
     */
    @Component("workerActor")
    @Scope("prototype")
    public class WorkerActor extends UntypedActor {
    
    	@Autowired
    	private BusinessService businessService;
    
    	private int count = 0;
    
    	@Override
    	public void onReceive(Object message) throws Exception {
    		if (message instanceof Request) {
    			businessService.perform(this + " " + (++count));
    		} else if (message instanceof Response) {
    			getSender().tell(count, getSelf());
    		} else {
    			unhandled(message);
    		}
    	}
    
    	public static class Request {
    	}
    
    	public static class Response {
    	}
    }

特别注意，使用依赖注入框架时，Actor不能配置为单例的，否则程序会出现问题。这里我们简单输出接受到的消息。

测试:

@Component
 class Runner implements CommandLineRunner {
 
 private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
 @Autowired
 private ActorSystem actorSystem;
 
 @Autowired
 private SpringExtension springExtension;
 
 @Override
 public void run(String[] args) throws Exception {
 try {
 	//Actor，它也是通过actorOf()方法创建
 ActorRef workerActor = actorSystem.actorOf(springExtension.props("workerActor"), "worker-actor");
 
 workerActor.tell(new WorkerActor.Request(), null);
 workerActor.tell(new WorkerActor.Request(), null);
 workerActor.tell(new WorkerActor.Request(), null);
 
 FiniteDuration duration = FiniteDuration.create(1, TimeUnit.SECONDS);
 Future<Object> awaitable = Patterns.ask(workerActor, new WorkerActor.Response(), Timeout.durationToTimeout(duration));
 
 logger.info("Response: " + Await.result(awaitable, duration));
 } finally {
 actorSystem.terminate();
 Await.result(actorSystem.whenTerminated(), Duration.Inf());
 }
 }
 }

启动测试结果：

08/08 18:49:19.246 INFO [main] i.k.s.s.SpringakkaApplication - Starting SpringakkaApplication on lihui with PID 61208 (D:\eclipse-workspace\workspace\springboot-akka\target\classes started by lily in D:\eclipse-workspace\workspace\springboot-akka)
    08/08 18:49:19.249 INFO [main] i.k.s.s.SpringakkaApplication - No active profile set, falling back to default profiles: default
    08/08 18:49:20.091 INFO [actor-system-akka.actor.default-dispatcher-4] a.e.s.Slf4jLogger - Slf4jLogger started
    08/08 18:49:20.296 INFO [main] i.k.s.s.SpringakkaApplication - Started SpringakkaApplication in 1.244 seconds (JVM running for 2.218)
    08/08 18:49:20.299 INFO [actor-system-akka.actor.default-dispatcher-5] i.k.s.s.s.BusinessService - Perform: io.kunalk.spring.springakka.actor.WorkerActor@70b82aef 1
    08/08 18:49:20.300 INFO [actor-system-akka.actor.default-dispatcher-5] i.k.s.s.s.BusinessService - Perform: io.kunalk.spring.springakka.actor.WorkerActor@70b82aef 2
    08/08 18:49:20.300 INFO [actor-system-akka.actor.default-dispatcher-5] i.k.s.s.s.BusinessService - Perform: io.kunalk.spring.springakka.actor.WorkerActor@70b82aef 3
    08/08 18:49:20.300 INFO [main] i.k.s.s.Runner - Response: 3

例子地址：

[https://github.com/ieflex/springboot-akka][https_github.com_ieflex_springboot-akka]

[wKiom1WOKpHgGDoXAAFY8MuvOgA842.jpg]: /images/20211104/fa571f21e78b41c593ecb076c957ae22.png
[wKiom1WOKpHgGDoXAAFY8MuvOgA842.jpg 1]: http://s3.51cto.com/wyfs02/M00/6E/FD/wKiom1WOKpHgGDoXAAFY8MuvOgA842.jpg
[wKiom1WOKyPBDRjpAAGSeStUphQ651.jpg]: /images/20211104/5ad459ef89f4446199da7b1c8209e2b6.png
[wKiom1WOKyPBDRjpAAGSeStUphQ651.jpg 1]: http://s3.51cto.com/wyfs02/M00/6E/FD/wKiom1WOKyPBDRjpAAGSeStUphQ651.jpg
[https_github.com_ieflex_springboot-akka]: https://github.com/ieflex/springboot-akka