C++中的Actor模型是一种并发编程模型，它通过将计算单元封装为独立、可并发执行的Actor实例，来实现并发和消息传递。每个Actor都有自己的状态和行为，并且通过接收和发送消息来进行通信。Actor之间是完全隔离的，它们之间只能通过消息进行通信，从而避免了共享状态和显式锁的问题。
下面是一个简单的C++ Actor模型的示例实现：
#include <IOStream>
#include <queue>
#include <thread>
#include <functional>
#include <mutex>
#include <condition_variable>
class Actor {
public:
void run() {
while (!stop) {
std::function<void()> message;
{
std::unique_lock<std::mutex> lock(mutex);
condition.wAIt(lock, [this]() {
return stop || !messageQueue.empty();
});
if (stop && messageQueue.empty()) {
return;
}
message = std::move(messageQueue.front());
messageQueue.pop();
}
message();
}
}
template<typename F, typename... Args>
void send(F&& f, Args&&... args) {
{
std::lock_guard<std::mutex> lock(mutex);
messageQueue.emplace(std::bind(std::forward<F>(f), std::forward<Args>(args)...));
}
condition.notify_one();
}
void stopActor() {
{
std::lock_guard<std::mutex> lock(mutex);
stop = true;
}
condition.notify_all();
}
private:
std::queue<std::function<void()>> messageQueue;
std::mutex mutex;
std::condition_variable condition;
bool stop = false;
};
// 示例Actor行为
class MyActor {
public:
void doSomething(int id) {
std::cout << "Actor ID: " << id << ", thread ID: " << std::this_thread::get_id() << std::endl;
// 具体行为逻辑
}
};
int main() {
MyActor actor;
Actor actingThread;
actingThread.send([&actor]() {
actor.doSomething(1);
});
actingThread.send([&actor]() {
actor.doSomething(2);
});
actingThread.send([&actingThread]() {
actingThread.stopActor();
});
std::thread thread([&actingThread]() {
actingThread.run();
});
thread.join();
return 0;
}
上述代码实现了一个简单的Actor模型。在Actor类中，使用条件变量和互斥锁来实现Actor的同步和通信。每个Actor都维护一个消息队列，通过send方法向队列中添加消息。
在main函数中，示例代码创建了一个MyActor对象并实例化了一个Actor对象。通过调用Actor对象的send方法，将消息发送给MyActor对象。MyActor对象中的doSomething方法会在调用线程的上下文中执行。
然后，通过调用stopActor方法停止Actor线程。最后，创建一个新线程运行Actor的run方法来处理消息队列中的消息。
请注意，上述示例代码只是一个简单的Actor模型的实现，实际使用时可能需要考虑更多的细节，如Actor之间的连接、创建多个Actor对象、消息路由等。Actor模型可以作为实现并发和并行的一种有趣和强大的方式，它可以提供更好的抽象和隔离性来处理并发编程中的问题。
要实现Actor之间的连接功能，可以使用一个中央调度器来管理Actor之间的消息传递和通信。下面是一个示例的C++ Actor模型的实现，包括Actor之间的连接功能：
#include <iostream>
#include <queue>
#include <thread>
#include <functional>
#include <mutex>
#include <condition_variable>
#include <unordered_map>
class Actor {
public:
void run() {
while (!stop) {
std::function<void()> message;
{
std::unique_lock<std::mutex> lock(mutex);
condition.wait(lock, [this]() {
return stop || !messageQueue.empty();
});
if (stop && messageQueue.empty()) {
return;
}
message = std::move(messageQueue.front());
messageQueue.pop();
}
message();
}
}
template<typename F, typename... Args>
void send(F&& f, Args&&... args) {
{
std::lock_guard<std::mutex> lock(mutex);
messageQueue.emplace(std::bind(std::forward<F>(f), std::forward<Args>(args)...));
}
condition.notify_one();
}
void stopActor() {
{
std::lock_guard<std::mutex> lock(mutex);
stop = true;
}
condition.notify_all();
}
private:
std::queue<std::function<void()>> messageQueue;
std::mutex mutex;
std::condition_variable condition;
bool stop = false;
};
class ActorSystem {
public:
ActorSystem() {
std::thread thread([this]() {
this->run();
});
thread.detach();
}
template<typename T>
Actor& createActor() {
std::unique_lock<std::mutex> lock(mutex);
Actor* actor = new Actor();
actors.emplace(actorId, actor);
actorId++;
return *actor;
}
void sendMessage(int actorId, std::function<void()> message) {
std::unique_lock<std::mutex> lock(mutex);
if (actors.find(actorId) != actors.end()) {
actors[actorId]->send(message);
}
}
void stopActor(int actorId) {
std::unique_lock<std::mutex> lock(mutex);
if (actors.find(actorId) != actors.end()) {
actors[actorId]->stopActor();
delete actors[actorId];
actors.erase(actorId);
}
}
private:
void run() {
while (true) {
std::this_thread::sleep_for(std::chrono::milliseconds(100));
std::unique_lock<std::mutex> lock(mutex);
for (const auto& actor : actors) {
actor.second->run();
}
}
}
std::unordered_map<int, Actor*> actors;
int actorId = 0;
std::mutex mutex;
};
// 示例Actor行为
class MyActor {
public:
MyActor(ActorSystem& actorSystem) : actorSystem(actorSystem) {}
void doSomething(int id) {
std::cout << "Actor ID: " << id << ", thread ID: " << std::this_thread::get_id() << std::endl;
// 具体行为逻辑
// 发送消息给其他Actor
actorSystem.sendMessage(1, [this]() {
this->doSomethingElse();
});
}
void doSomethingElse() {
std::cout << "Doing something else" << std::endl;
// 具体行为逻辑
}
private:
ActorSystem& actorSystem;
};
int main() {
ActorSystem actorSystem;
MyActor actor1(actorSystem);
MyActor actor2(actorSystem);
Actor& actor1Ref = actorSystem.createActor<MyActor>();
Actor& actor2Ref = actorSystem.createActor<MyActor>();
actor1Ref.send([&actor1]() {
actor1.doSomething(1);
});
actor2Ref.send([&actor2]() {
actor2.doSomething(2);
});
std::this_thread::sleep_for(std::chrono::seconds(5));
actorSystem.stopActor(0);
actorSystem.stopActor(1);
return 0;
}
上述代码中，我们引入了一个ActorSystem类来管理所有的Actor。ActorSystem类中维护了一个unordered_map来存储Actor的ID和指针的映射关系。createActor方法用于创建新的Actor，并返回其引用。sendMessage方法用于向指定的Actor发送消息，stopActor方法用于停止指定的Actor。
在MyActor类中，我们添加了一个doSomethingElse方法，用于演示如何向其他Actor发送消息。在doSomething方法中，我们通过actorSystem.sendMessage方法向ID为1的Actor发送消息。
在main函数中，我们创建了两个MyActor对象，并通过createActor方法创建了两个对应的Actor。然后，我们通过调用send方法向两个Actor发送消息。最后，我们通过调用stopActor方法停止两个Actor。
请注意，上述示例代码只是一个简单的Actor模型的实现，实际使用时可能需要考虑更多的细节，如消息路由、Actor之间的连接关系、消息队列的优化等。Actor模型是一种强大的并发编程模型，可以提供更好的抽象和隔离性来处理并发编程中的问题。
要添加消息路由功能，我们可以为ActorSystem类添加一个Router类成员变量，并在sendMessage方法中使用路由器来选择目标Actor。下面是更新后的代码：
class Router {
public:
virtual int route(const std::unordered_map<int, Actor*>& actors) = 0;
};
class RoundRobinRouter : public Router {
public:
int route(const std::unordered_map<int, Actor*>& actors) override {
if (actors.empty()) {
return -1;
}
currentActorIndex = (currentActorIndex + 1) % actors.size();
auto it = std::next(actors.begin(), currentActorIndex);
return it->first;
}
private:
int currentActorIndex = -1;
};
class ActorSystem {
public:
ActorSystem() {
std::thread thread([this]() {
this->run();
});
thread.detach();
}
template<typename T>
Actor& createActor() {
std::unique_lock<std::mutex> lock(mutex);
Actor* actor = new Actor();
actors.emplace(actorId, actor);
actorId++;
return *actor;
}
void sendMessage(std::function<void()> message, Router* router = nullptr) {
std::unique_lock<std::mutex> lock(mutex);
int actorId = -1;
if (router != nullptr) {
actorId = router->route(actors);
} else {
if (!actors.empty()) {
actorId = actors.begin()->first;
}
}
if (actorId != -1) {
actors[actorId]->send(message);
}
}
void stopActor(int actorId) {
std::unique_lock<std::mutex> lock(mutex);
if (actors.find(actorId) != actors.end()) {
actors[actorId]->stopActor();
delete actors[actorId];
actors.erase(actorId);
}
}
private:
void run() {
while (true) {
std::this_thread::sleep_for(std::chrono::milliseconds(100));
std::unique_lock<std::mutex> lock(mutex);
for (const auto& actor : actors) {
actor.second->run();
}
}
}
std::unordered_map<int, Actor*> actors;
int actorId = 0;
std::mutex mutex;
};
在上述代码中，我们添加了一个Router类和一个具体的RoundRobinRouter类。Router类是一个抽象类，其中定义了一个route方法，用于根据一组Actor选择目标Actor。RoundRobinRouter类是一个具体的路由器实现，它使用轮询的方式选择目标Actor。
在ActorSystem类中，我们添加了一个可选的路由器参数router，并在sendMessage方法中使用路由器来选择目标Actor。如果没有指定路由器，则默认选择第一个Actor。
你可以根据需要实现其他的路由器策略，比如随机选择、基于消息负载的选择等。
使用示例代码：
int main() {
ActorSystem actorSystem;
MyActor actor1(actorSystem);
MyActor actor2(actorSystem);
Actor& actor1Ref = actorSystem.createActor<MyActor>();
Actor& actor2Ref = actorSystem.createActor<MyActor>();
RoundRobinRouter router;
actorSystem.sendMessage([&actor1]() {
actor1.doSomething(1);
}, &router);
actorSystem.sendMessage([&actor2]() {
actor2.doSomething(2);
}, &router);
std::this_thread::sleep_for(std::chrono::seconds(5));
actorSystem.stopActor(0);
actorSystem.stopActor(1);
return 0;
}
在上述示例代码中，我们创建了一个RoundRobinRouter对象，并将其作为参数传递给sendMessage方法。这样，消息将根据轮询策略路由到不同的Actor上。
请注意，上述示例代码仅仅是一个简单的消息路由实现，实际使用时可能需要考虑更多的细节，比如路由策略的灵活性、动态添加/删除Actor时的路由更新等。
要优化路由策略的灵活性，我们可以将路由策略从Router类中分离出来，并使用策略模式来实现。下面是更新后的代码：
class RoutingStrategy {
public:
virtual int selectActor(const std::unordered_map<int, Actor*>& actors) = 0;
};
class Router {
public:
Router(RoutingStrategy* strategy) : strategy(strategy) {}
int route(const std::unordered_map<int, Actor*>& actors) {
return strategy->selectActor(actors);
}
private:
RoutingStrategy* strategy;
};
class RoundRobinStrategy : public RoutingStrategy {
public:
int selectActor(const std::unordered_map<int, Actor*>& actors) override {
if (actors.empty()) {
return -1;
}
currentActorIndex = (currentActorIndex + 1) % actors.size();
auto it = std::next(actors.begin(), currentActorIndex);
return it->first;
}
private:
int currentActorIndex = -1;
};
class RandomStrategy : public RoutingStrategy {
public:
int selectActor(const std::unordered_map<int, Actor*>& actors) override {
if (actors.empty()) {
return -1;
}
std::random_device rd;
std::mt19937 gen(rd());
std::uniform_int_distribution<> dis(0, actors.size() - 1);
auto it = std::next(actors.begin(), dis(gen));
return it->first;
}
};
class ActorSystem {
public:
ActorSystem() {
std::thread thread([this]() {
this->run();
});
thread.detach();
}
template<typename T>
Actor& createActor() {
std::unique_lock<std::mutex> lock(mutex);
Actor* actor = new Actor();
actors.emplace(actorId, actor);
actorId++;
return *actor;
}
void sendMessage(std::function<void()> message, RoutingStrategy* strategy = nullptr) {
std::unique_lock<std::mutex> lock(mutex);
int actorId = -1;
if (strategy != nullptr) {
actorId = strategy->selectActor(actors);
} else {
if (!actors.empty()) {
actorId = actors.begin()->first;
}
}
if (actorId != -1) {
actors[actorId]->send(message);
}
}
void stopActor(int actorId) {
std::unique_lock<std::mutex> lock(mutex);
if (actors.find(actorId) != actors.end()) {
actors[actorId]->stopActor();
delete actors[actorId];
actors.erase(actorId);
}
}
private:
void run() {
while (true) {
std::this_thread::sleep_for(std::chrono::milliseconds(100));
std::unique_lock<std::mutex> lock(mutex);
for (const auto& actor : actors) {
actor.second->run();
}
}
}
std::unordered_map<int, Actor*> actors;
int actorId = 0;
std::mutex mutex;
};
在上述代码中，我们将路由策略从Router类中分离出来，并定义了一个RoutingStrategy抽象类。RoutingStrategy类中定义了一个selectActor方法，用于根据一组Actor选择目标Actor。
然后，我们实现了两个具体的路由策略类：RoundRobinStrategy和RandomStrategy。RoundRobinStrategy类使用轮询的方式选择目标Actor，RandomStrategy类使用随机选择的方式选择目标Actor。
在Router类中，我们使用了策略模式，将具体的路由策略对象存储在成员变量中，并在route方法中调用策略对象的selectActor方法。
在ActorSystem类中，我们添加了一个可选的策略参数strategy，并在sendMessage方法中使用策略对象来选择目标Actor。如果没有指定策略对象，则默认选择第一个Actor。
使用示例代码：
int main() {
ActorSystem actorSystem;
MyActor actor1(actorSystem);
MyActor actor2(actorSystem);
Actor& actor1Ref = actorSystem.createActor<MyActor>();
Actor& actor2Ref = actorSystem.createActor<MyActor>();
RoundRobinStrategy roundRobinStrategy;
RandomStrategy randomStrategy;
Router roundRobinRouter(&roundRobinStrategy);
Router randomRouter(&randomStrategy);
actorSystem.sendMessage([&actor1]() {
actor1.doSomething(1);
}, &roundRobinRouter);
actorSystem.sendMessage([&actor2]() {
actor2.doSomething(2);
}, &randomRouter);
std::this_thread::sleep_for(std::chrono::seconds(5));
actorSystem.stopActor(0);
actorSystem.stopActor(1);
return 0;
}
在上述示例代码中，我们创建了一个RoundRobinStrategy对象和一个RandomStrategy对象，并将它们分别传递给RoundRobinRouter和RandomRouter对象。这样，消息将根据不同的策略路由到不同的Actor上。
通过使用策略模式，我们可以很容易地添加新的路由策略，只需实现一个新的策略类，并将其传递给Router对象即可。这样，我们可以根据需要灵活地选择不同的路由策略。
当动态添加或删除Actor时，需要更新路由策略，以确保新添加的Actor也能够接收到消息。
对于动态添加Actor，我们可以在ActorSystem类中添加一个addActor方法，用于向actors容器中添加新的Actor。在添加完Actor之后，我们需要更新路由策略对象中的Actor列表。
对于动态删除Actor，我们可以在ActorSystem类中添加一个removeActor方法，用于从actors容器中删除指定的Actor。在删除完Actor之后，我们同样需要更新路由策略对象中的Actor列表。
下面是更新后的ActorSystem类的代码：
class ActorSystem {
public:
ActorSystem() {
std::thread thread([this]() {
this->run();
});
thread.detach();
}
template<typename T>
Actor& createActor() {
std::unique_lock<std::mutex> lock(mutex);
Actor* actor = new Actor();
actors.emplace(actorId, actor);
actorId++;
// Update routing strategy
for (auto& strategy : routingStrategies) {
strategy->updateActors(actors);
}
return *actor;
}
void addActor(int actorId, Actor* actor) {
std::unique_lock<std::mutex> lock(mutex);
actors.emplace(actorId, actor);
// Update routing strategy
for (auto& strategy : routingStrategies) {
strategy->updateActors(actors);
}
}
void removeActor(int actorId) {
std::unique_lock<std::mutex> lock(mutex);
if (actors.find(actorId) != actors.end()) {
actors.erase(actorId);
// Update routing strategy
for (auto& strategy : routingStrategies) {
strategy->updateActors(actors);
}
}
}
void sendMessage(std::function<void()> message, RoutingStrategy* strategy = nullptr) {
std::unique_lock<std::mutex> lock(mutex);
int actorId = -1;
if (strategy != nullptr) {
actorId = strategy->selectActor(actors);
} else {
if (!actors.empty()) {
actorId = actors.begin()->first;
}
}
if (actorId != -1) {
actors[actorId]->send(message);
}
}
void stopActor(int actorId) {
std::unique_lock<std::mutex> lock(mutex);
if (actors.find(actorId) != actors.end()) {
actors[actorId]->stopActor();
delete actors[actorId];
actors.erase(actorId);
}
}
void addRoutingStrategy(RoutingStrategy* strategy) {
routingStrategies.push_back(strategy);
strategy->updateActors(actors);
}
private:
void run() {
while (true) {
std::this_thread::sleep_for(std::chrono::milliseconds(100));
std::unique_lock<std::mutex> lock(mutex);
for (const auto& actor : actors) {
actor.second->run();
}
}
}
std::unordered_map<int, Actor*> actors;
std::vector<RoutingStrategy*> routingStrategies;
int actorId = 0;
std::mutex mutex;
};
在更新后的ActorSystem类中，我们添加了一个routingStrategies成员变量，用于存储所有的路由策略对象。在添加或删除Actor时，我们需要遍历routingStrategies并调用每个策略对象的updateActors方法，以更新Actor列表。
此外，我们还添加了一个addRoutingStrategy方法，用于向routingStrategies中添加新的路由策略对象。在添加完路由策略对象之后，我们同样需要调用updateActors方法来更新Actor列表。
使用示例代码：
int main() {
ActorSystem actorSystem;
MyActor actor1(actorSystem);
MyActor actor2(actorSystem);
Actor& actor1Ref = actorSystem.createActor<MyActor>();
Actor& actor2Ref = actorSystem.createActor<MyActor>();
RoundRobinStrategy roundRobinStrategy;
RandomStrategy randomStrategy;
actorSystem.addRoutingStrategy(&roundRobinStrategy);
actorSystem.addRoutingStrategy(&randomStrategy);
actorSystem.sendMessage([&actor1]() {
actor1.doSomething(1);
}, &roundRobinStrategy);
actorSystem.sendMessage([&actor2]() {
actor2.doSomething(2);
}, &randomStrategy);
std::this_thread::sleep_for(std::chrono::seconds(5));
actorSystem.stopActor(0);
actorSystem.stopActor(1);
return 0;
}
在上述示例代码中，我们首先创建了两个MyActor对象，并将它们添加到ActorSystem中。然后，我们创建了一个RoundRobinStrategy对象和一个RandomStrategy对象，并将它们分别添加到ActorSystem中。
接下来，我们使用sendMessage方法向不同的路由策略发送消息。在发送消息之前，我们需要将路由策略对象传递给sendMessage方法。
最后，我们停止了两个Actor，并结束了程序。在停止Actor之后，我们需要调用stopActor方法，并在ActorSystem中删除相应的Actor。
#include <iostream>
#include <thread>
#include <chrono>
#include <unordered_map>
#include <functional>
#include <mutex>
class Actor {
public:
Actor() {}
void send(std::function<void()> message) {
messages.push_back(message);
}
void run() {
std::unique_lock<std::mutex> lock(mutex);
for (const auto& message : messages) {
message();
}
messages.clear();
}
void stopActor() {
std::unique_lock<std::mutex> lock(mutex);
stopped = true;
}
private:
std::vector<std::function<void()>> messages;
std::mutex mutex;
bool stopped = false;
};
class RoutingStrategy {
public:
virtual int selectActor(const std::unordered_map<int, Actor*>& actors) = 0;
virtual void updateActors(const std::unordered_map<int, Actor*>& actors) = 0;
};
class RoundRobinStrategy : public RoutingStrategy {
public:
int selectActor(const std::unordered_map<int, Actor*>& actors) override {
if (actors.empty()) {
return -1;
}
if (currentActor >= actors.size()) {
currentActor = 0;
}
auto it = actors.begin();
std::advance(it, currentActor);
currentActor++;
return it->first;
}
void updateActors(const std::unordered_map<int, Actor*>& actors) override {
this->actors = actors;
}
private:
std::unordered_map<int, Actor*> actors;
int currentActor = 0;
};
class RandomStrategy : public RoutingStrategy {
public:
int selectActor(const std::unordered_map<int, Actor*>& actors) override {
if (actors.empty()) {
return -1;
}
int randomIndex = rand() % actors.size();
auto it = actors.begin();
std::advance(it, randomIndex);
return it->first;
}
void updateActors(const std::unordered_map<int, Actor*>& actors) override {
this->actors = actors;
}
private:
std::unordered_map<int, Actor*> actors;
};
class ActorSystem {
public:
template<typename ActorType>
ActorType& createActor() {
std::unique_lock<std::mutex> lock(mutex);
ActorType* actor = new ActorType(*this);
actors.emplace(actorId, actor);
actorId++;
// Update routing strategy
for (auto& strategy : routingStrategies) {
strategy->updateActors(actors);
}
return *actor;
}
void addActor(int actorId, Actor* actor) {
std::unique_lock<std::mutex> lock(mutex);
actors.emplace(actorId, actor);
// Update routing strategy
for (auto& strategy : routingStrategies) {
strategy->updateActors(actors);
}
}
void removeActor(int actorId) {
std::unique_lock<std::mutex> lock(mutex);
if (actors.find(actorId) != actors.end()) {
actors.erase(actorId);
// Update routing strategy
for (auto& strategy : routingStrategies) {
strategy->updateActors(actors);
}
}
}
void sendMessage(std::function<void()> message, RoutingStrategy* strategy = nullptr) {
std::unique_lock<std::mutex> lock(mutex);
int actorId = -1;
if (strategy != nullptr) {
actorId = strategy->selectActor(actors);
} else {
if (!actors.empty()) {
actorId = actors.begin()->first;
}
}
if (actorId != -1) {
actors[actorId]->send(message);
}
}
void stopActor(int actorId) {
std::unique_lock<std::mutex> lock(mutex);
if (actors.find(actorId) != actors.end()) {
actors[actorId]->stopActor();
delete actors[actorId];
actors.erase(actorId);
}
}
void addRoutingStrategy(RoutingStrategy* strategy) {
routingStrategies.push_back(strategy);
strategy->updateActors(actors);
}
private:
void run() {
while (true) {
std::this_thread::sleep_for(std::chrono::milliseconds(100));
std::unique_lock<std::mutex> lock(mutex);
for (const auto& actor : actors) {
actor.second->run();
}
}
}
std::unordered_map<int, Actor*> actors;
std::vector<RoutingStrategy*> routingStrategies;
int actorId = 0;
std::mutex mutex;
};
class MyActor {
public:
MyActor(ActorSystem& actorSystem) : actorSystem(actorSystem) {
actorSystem.addActor(actorId, this);
}
void doSomething(int value) {
std::cout << "Actor " << actorId << " is doing something with value " << value << std::endl;
}
private:
int actorId;
ActorSystem& actorSystem;
};
int main() {
ActorSystem actorSystem;
MyActor actor1(actorSystem);
MyActor actor2(actorSystem);
Actor& actor1Ref = actorSystem.createActor<MyActor>();
Actor& actor2Ref = actorSystem.createActor<MyActor>();
RoundRobinStrategy roundRobinStrategy;
RandomStrategy randomStrategy;
actorSystem.addRoutingStrategy(&roundRobinStrategy);
actorSystem.addRoutingStrategy(&randomStrategy);
actorSystem.sendMessage([&actor1]() {
actor1.doSomething(1);
}, &roundRobinStrategy);
actorSystem.sendMessage([&actor2]() {
actor2.doSomething(2);
}, &randomStrategy);
std::this_thread::sleep_for(std::chrono::seconds(5));
actorSystem.stopActor(0);
actorSystem.stopActor(1);
return 0;
}
京公网安备 11010802035086号