26

我正在尝试学习 Akka 演员和期货,但在阅读了http://akka.io上的文档 并做了http://doc.akka.io/docs/akka/2.0.2/intro/getting-started-first -java.html 我仍然有一些理解问题。我想计算 Pi 的值也是很多人都可以理解的,但不是我 =)。我搜索了一下,但没有找到任何适合我的例子。因此,我想我会拿一些我的真实代码并把它扔在这里,并用它来交换一个如何用 Akka 做到这一点的例子。

好的,我们开始吧:

我有一个 java play2 应用程序,我需要从我的数据库中获取一些数据并将其索引到我的 elasticsearch 实例中。

  1. 我打电话给 DB 并获取场地的 ID。

  2. 然后我拆分列表并创建几个可调用的索引任务。

  3. 之后,我调用所有任务,其中每个任务从数据库中收集分配的 id 的场所。

  4. 对于每个场所,将其索引到 elasticsearch 实例并使其可搜索。

  5. 完毕。

应用程序.java:

public class Application extends Controller {

  private static final int VENUE_BATCH = 1000;
  private static int size;

  public static Result index() {

      List<Long> venueIds = DbService.getAllVenueIds();
      size = venueIds.size();
      Logger.info("Will index " + size + " items in total.");
      ExecutorService service = Executors.newFixedThreadPool(getRuntime().availableProcessors());
      int startIx = 0;
      Collection<Callable<Object>> indexTasks = new ArrayList<Callable<Object>>();
      do {
          int endIx = Math.min(startIx + VENUE_BATCH, size);
          List<Long> subList = venueIds.subList(startIx, endIx);
          VenueIndexTask indexTask = new VenueIndexTask(subList);
          indexTasks.add(indexTask);
      } while ((startIx += VENUE_BATCH) < size);

      Logger.info("Invoking all tasks!");
      try {
          service.invokeAll(indexTasks);
      } catch (InterruptedException e) {
          e.printStackTrace();
      }

      return ok(index.render("Done indexing."));
  } 
}

场地任务:

public class VenueIndexTask implements Callable<Object> {

    private List<Long> idSubList;

    public VenueIndexTask(List<Long> idSubList){
        this.idSubList = idSubList;
        Logger.debug("Creating task which will index " + idSubList.size() + " items. " +
                "Range: " + rangeAsString() + ".");
    }

    @Override
    public Object call() throws Exception {
        List<Venue> venues = DbService.getVenuesForIds(idSubList);
        Logger.debug("Doing some indexing: "+venues.size());

        for(Venue venue : venues) {
            venue.index();
        }
        return null;
    }
    private String rangeAsString() {
        return "[" + idSubList.get(0) + "-" + idSubList.get(idSubList.size() - 1) + "]";
    }
}

地点:

@IndexType(name = "venue")
public class Venue extends Index {

    private String name;

    // Find method static for request
    public static Finder<Venue> find = new Finder<Venue>(Venue.class);

    public Venue() {
    }

    public Venue(String id, String name) {
        super.id = id;
        this.name = name;
    }

    @Override
    public Map toIndex() {
        HashMap map = new HashMap();
        map.put("id", super.id);
        map.put("name", name);
        return map;
    }

    @Override
    public Indexable fromIndex(Map map) {
        if (map == null) {
            return this;
        }
        this.name = (String) map.get("name");
        return this;
    }
}

所以你们所有的阿卡人都发疯了!请尽可能多地提出可以使用的酷期货功能或我可以用来学习这些东西的任何其他知识/代码。

4

2 回答 2

22

我对 Akka(或任何其他基于消息的系统)的看法是像传送带一样思考,就像在工厂中一样。Actors 中一种简化的思维方式可能是接受披萨订单。

  • 你,饥饿的顾客(演员/角色)向披萨店发送订单(A Message)

  • 客服(演员/角色)接您的订单,给您订单号(未来)

  • 如果您不耐烦,您可能会在电话/互联网/商店等到您拿到披萨(同步/阻塞交易),否则您会对订单号感到满意,稍后再检查(非阻塞)

  • 客服在厨房经理(演员)的监督下将消息发送给厨师(演员)。这是一个流程很重的厨房,有层次感。阿卡喜欢这样。见监督

  • Chef 创建了一个新的 Pizza 并附加了订单的详细信息(一条新消息),然后通过送货经理(Supervisor Actor)将其传递给送货员(Actor)。

  • 在此过程中,您的订单详细信息没有更改,那将是一场噩梦。如果你想要纯奶酪,如果你有意大利辣香肠,你会不高兴!所有消息都应该是不可变的!但是,对于不同的参与者,消息可能会有所不同。送货员会期待披萨和附上的订单详细信息,厨师会期待订单。当消息需要更改时,会创建一条新消息。

  • 每个演员都擅长一个角色,如果一个人必须完成所有任务,效果如何?可能是某些参与者的数量超过了其他参与者(厨师 10 个线程,送货员 2 个线程,客户服务 1 个线程)。

  • 阻止行为是一种痛苦,想象一下客户服务正在等待厨师和送货员在看到下一位客户之前?

希望我对您有所帮助,这是一个巨大的话题和巨大的思想转变。祝你好运

于 2014-01-07T13:52:54.697 回答
1

Coursera 目前开设了一门关于反应式编程的课程,其中有最后 3 堂课是关于 Akka 和 Actor 模型的。这包括视频讲座和家庭作业(在 Scala 中,但不是 Java)。虽然您收到完整证书为时已晚,但您仍然可以参加课程并检查最后三周。

https://class.coursera.org/reactive-001/class

于 2013-12-19T08:54:46.247 回答