import org.apache.hadoop.yarn.server.resourcemanager.Task.State; //导入依赖的package包/类
private synchronized void assign(Priority priority, NodeType type, 
    List<Container> containers) throws IOException, YarnException {
  for (Iterator<Container> i=containers.iterator(); i.hasNext();) {
    Container container = i.next();
    String host = container.getNodeId().toString();
    
    if (Resources.equals(requestSpec.get(priority), container.getResource())) { 
      // See which task can use this container
      for (Iterator<Task> t=tasks.get(priority).iterator(); t.hasNext();) {
        Task task = t.next();
        if (task.getState() == State.PENDING && task.canSchedule(type, host)) {
          NodeManager nodeManager = getNodeManager(host);
          
          task.start(nodeManager, container.getId());
          i.remove();
          
          // Track application resource usage
          Resources.addTo(used, container.getResource());
          
          LOG.info("Assigned container (" + container + ") of type " + type +
              " to task " + task.getTaskId() + " at priority " + priority + 
              " on node " + nodeManager.getHostName() +
              ", currently using " + used + " resources");

          // Update resource requests
          updateResourceRequests(requests.get(priority), type, task);

          // Launch the container
          StartContainerRequest scRequest =
              StartContainerRequest.newInstance(createCLC(),
                container.getContainerToken());
          List<StartContainerRequest> list =
              new ArrayList<StartContainerRequest>();
          list.add(scRequest);
          StartContainersRequest allRequests =
              StartContainersRequest.newInstance(list);
          nodeManager.startContainers(allRequests);
          break;
        }
      }
    }
  }
}