问题描述
我只是在玩一点 Actix web 和 Actix actor，当我构建一个简单的应用程序时，我观察到 Actix 正在创建多个 actor。我想限制只有一个，但我做不到。
我在这里做错了什么。我无法在闭包之外构建 Actor Addr 并在路由中使用，因为它会导致编译错误。但是如果我在闭包内移动 actor 的创建，它会被创建不止一次，这取决于 worker 的数量。
我想要实现的是，不管有多少工人，我只需要使用一个演员。
main.rs
mod millis_provider;
use actix::*;
use actix_web::{App,get,HttpResponse,HttpServer,Responder,web};
use crate::millis_provider::Millis;
use crate::millis_provider::MillisProvider;
struct MillisConsumer {
millis_provider: Addr<MillisProvider>
}
impl MillisConsumer {
fn new(millis_provider: Addr<MillisProvider>) -> Self {
MillisConsumer { millis_provider }
}
async fn millis(&self) -> std::result::Result<u128,std::io::Error> {
self.millis_provider.send(Millis {}).await.expect("Failed to get result from actor")
}
}
#[get("/")]
async fn hello() -> impl Responder {
HttpResponse::Ok().body("Use /millis GET request to obtain unique millis")
}
#[get("/millis")]
async fn millis(millis_consumer: web::Data<MillisConsumer>) -> impl Responder {
let millis = &millis_consumer.millis().await.unwrap();
HttpResponse::Ok().body(millis.to_string())
}
use structopt::StructOpt;
use simple_logger::SimpleLogger;
#[derive(StructOpt)]
struct GeneratorConfig {
http_host: String,http_port: i32,}
#[actix_web::main]
async fn main() -> std::io::Result<()> {
let config = GeneratorConfig::from_args();
SimpleLogger::new().init().unwrap();
// let actor: Addr<MillisProvider> =
// MillisProvider {
// current_millis: 0,// total_millis_generated: 0,// }.start();
HttpServer::new(move || {
let actor: Addr<MillisProvider> =
MillisProvider {
current_millis: 0,total_millis_generated: 0,}.start();
App::new()
.data(MillisConsumer::new(actor))
.service(millis)
.service(hello)
})
// .workers(1)
.bind(format!("{}:{}",config.http_host,config.http_port))?
.run()
.await
}
millis_provider
use log::{info};
use std::time::{SystemTime,UNIX_EPOCH};
use actix::{Actor,Handler,Message,Context};
#[derive(Message)]
#[rtype(result = "std::io::Result<u128>")]
pub(crate) struct Millis;
pub(crate) struct MillisProvider {
pub(crate) current_millis: u128,pub(crate) total_millis_generated: u128,}
fn get_epoch_ms() -> u128 {
SystemTime::Now()
.duration_since(UNIX_EPOCH)
.map(|x| x.as_millis())
.expect("Failed to get time in millis")
}
impl Actor for MillisProvider {
type Context = Context<Self>;
fn started(&mut self,_: &mut Self::Context) {
info!("Actor started");
}
fn stopped(&mut self,_: &mut Self::Context) {
info!("Actor is stopped");
}
}
impl Handler<Millis> for MillisProvider {
type Result = std::io::Result<u128>;
fn handle(&mut self,_: Millis,_: &mut Context<Self>) -> Self::Result {
let mut current_time = get_epoch_ms();
while current_time == self.current_millis {
current_time = get_epoch_ms()
}
self.current_millis = current_time;
self.total_millis_generated += 1;
if self.total_millis_generated % 1000 == 0 {
info!("Generated total of {} ids",self.total_millis_generated);
}
Ok(self.current_millis)
}
}
Cargo.toml
[package]
name = "rust-actix-example"
version = "0.1.0"
authors = ["mtekp <manjunath@gmail.com>"]
edition = "2018"
[dependencies]
log = "0.4"
structopt = "0.3.13"
actix = "0.10"
actix-web = "3"
simple_logger = "1.11"
日志
2021-03-03 20:43:12,090 INFO [actix_server::builder] Starting 4 workers
2021-03-03 20:43:12,091 TRACE [mio::poll] registering with poller
2021-03-03 20:43:12,091 INFO [actix_server::builder] Starting "actix-web-service-127.0.0.1:8080" service on 127.0.0.1:8080
2021-03-03 20:43:12,092 TRACE [mio::poll] registering with poller
2021-03-03 20:43:12,092 INFO [rust_actix_example::millis_provider] Actor started
2021-03-03 20:43:12,092 TRACE [actix_server::worker] Service "actix-web-service-127.0.0.1:8080" is available
2021-03-03 20:43:12,093 TRACE [actix_server::worker] Service "actix-web-service-127.0.0.1:8080" is available
2021-03-03 20:43:12,093 TRACE [mio::poll] registering with poller
2021-03-03 20:43:12,093 INFO [rust_actix_example::millis_provider] Actor started
2021-03-03 20:43:12,094 TRACE [mio::poll] registering with poller
2021-03-03 20:43:12,094 TRACE [actix_server::worker] Service "actix-web-service-127.0.0.1:8080" is available
根据工作人员的数量，我们可以在这里看到 Actor started 4 次。
和cargo run -- 127.0.0.1 8080
解决方法
有效的更改是，为 Clone 派生 MillisConsumer 并将其克隆到闭包中。
新的 main.rs
mod millis_provider;
use actix::*;
use actix_web::{App,get,HttpResponse,HttpServer,Responder,web};
use crate::millis_provider::Millis;
use crate::millis_provider::MillisProvider;
#[derive(Clone)]
struct MillisConsumer {
millis_provider: Addr<MillisProvider>
}
impl MillisConsumer {
fn new(millis_provider: Addr<MillisProvider>) -> Self {
MillisConsumer { millis_provider }
}
async fn millis(&self) -> std::result::Result<u128,std::io::Error> {
self.millis_provider.send(Millis {}).await.expect("Failed to get result from actor")
}
}
#[get("/")]
async fn hello() -> impl Responder {
HttpResponse::Ok().body("Use /millis GET request to obtain unique millis")
}
#[get("/millis")]
async fn millis(millis_consumer: web::Data<MillisConsumer>) -> impl Responder {
let millis = &millis_consumer.millis().await.unwrap();
HttpResponse::Ok().body(millis.to_string())
}
use structopt::StructOpt;
use simple_logger::SimpleLogger;
#[derive(StructOpt)]
struct GeneratorConfig {
http_host: String,http_port: i32,}
#[actix_web::main]
async fn main() -> std::io::Result<()> {
let config = GeneratorConfig::from_args();
SimpleLogger::new().init().unwrap();
let actor: Addr<MillisProvider> =
MillisProvider {
current_millis: 0,total_millis_generated: 0,}.start();
let consumer = MillisConsumer::new(actor);
HttpServer::new(move || {
let c = consumer.clone();
App::new()
.data(c)
.service(millis)
.service(hello)
})
.bind(format!("{}:{}",config.http_host,config.http_port))?
.run()
.await
}