标签: parallel-processing

问题:我想实现几个正在侦听MQ服务器队列以进行异步作业的php-worker进程.现在的问题是,简单地将这个进程作为守护进程在服务器上运行并不能真正给我任何级别的控制(负载,状态,锁定)......除了可能用于转储ps -aux.因此,我正在寻找一种允许我监视和控制实例的运行时环境,无论是在系统(进程)级别还是在更高层(某种Java风格的应用程序服务器)上

有什么指针吗？

我有这样的取消令牌

   static CancellationTokenSource TokenSource= new CancellationTokenSource();

我有一个像这样的阻止集合

BlockingCollection<object> items= new BlockingCollection<object>();

var item = items.Take(TokenSource.Token);

if(TokenSource.CancelPending)
   return;

我打电话的时候

TokenSource.Cancel();

Take不会像它应该的那样继续下去.如果我使用带轮询的TryTake,令牌显示它被设置为已取消.

每次我都要创建一个可变数量的线程.我这样做是通过创建一个Threads数组并创建多个线程.

但是,我不明白如何启动这些n个线程表现得像多线程概念.我希望他们并行运行.

请指导在这个场所做什么.

我试图使用新的并行功能JDK8,但不幸的是,我无法让它工作.NetBeans 7.1表示不存在"并行"方法.

这种方法需要特殊导入吗？有没有人有示例代码演示Java 8并行性？

我有4个批处理文件.我想运行one.bat,并two.bat在一次,兼任.这两个批处理文件完成后,three.bat并four.bat应同时运行,在并行.我试过很多方法,但mot工作得很好.

任何人都可以帮助我吗？

背景

我有一个应用程序接收定期数据转储(XML文件)并使用Entity Framework 5(代码优先)将它们导入现有数据库.导入通过EF5而不是BULK INSERT或BCP进行,因为必须应用已存在于实体中的业务规则.

处理似乎是应用程序本身的CPU绑定(极快,启用了写缓存的磁盘IO子系统在整个过程中显示几乎为零的磁盘等待时间,而SQL Server显示的CPU时间不超过8%-10%).

为了提高效率,我使用TPL Dataflow构建了一个管道,其组件包括:

Read & Parse XML file
        |
        V
Create entities from XML Node
        |
        V
Batch entities (BatchBlock, currently n=200)
        |
        V
Create new DbContext / insert batched entities / ctx.SaveChanges()

通过这样做,我看到性能大幅提升,但不能使CPU高于60%.

分析

怀疑某种资源争用,我使用VS2012 Profiler的资源争用数据(并发)模式运行该过程.

分析器显示52%的争用标记为句柄2的资源.钻进,我看到创建Handle 2最多争用的方法是

System.Data.Entity.Internal.InternalContext.SaveChanges()

第二名,与SaveChanges()的争论大约是40%

System.Data.Entity.DbSet`1.Add(!0)

问题

• 我怎样才能弄清楚Handle 2究竟是什么(例如TPL的一部分,EF的一部分)？
• EF限制调用将DbContext实例与单独的线程分开吗？似乎他们正在争夺一个共享资源.
• 在这种情况下,我有什么办法可以改善并行性吗？

UPDATE

对于有问题的运行,调用SaveChanges的任务的最大并行度设置为12(我尝试了各种值,包括先前运行中的Unbounded).

更新2

微软的EF团队提供了反馈意见.请参阅我的答案以获取摘要.

当流源是a时,我无法实现流处理的良好并行化Reader.在四核CPU上运行下面的代码我首先观察到3个核心,然后突然下降到两个核心,然后是一个核心.整体CPU利用率徘徊在50%左右.

请注意示例的以下特征:

• 只有6,000行;
• 每条线需要大约20毫秒来处理;
• 整个过程大约需要一分钟.

这意味着所有压力都在CPU上,I/O很小.这个例子是一个用于自动并行化的坐鸭.

import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;

... class imports elided ...    

public class Main
{
  static final AtomicLong totalTime = new AtomicLong();

  public static void main(String[] args) throws IOException {
    final long start = System.nanoTime();
    final Path inputPath = createInput();
    System.out.println("Start processing");

    try (PrintWriter w = new PrintWriter(Files.newBufferedWriter(Paths.get("output.txt")))) {
      Files.lines(inputPath).parallel().map(Main::processLine)
        .forEach(w::println);
    }

    final double cpuTime = totalTime.get(),
                 realTime = System.nanoTime()-start;
    final int cores = Runtime.getRuntime().availableProcessors();
    System.out.println("          Cores: " + cores);
    System.out.format("       CPU …

考虑以下:

import scala.concurrent._
import scala.concurrent.duration.Duration.Inf
import scala.concurrent.ExecutionContext.Implicits.global

def slowInt(i: Int) = { Thread.sleep(200); i }
def slowAdd(x: Int, y: Int) = { Thread.sleep(100); x + y }
def futures = (1 to 20).map(i => future(slowInt(i)))

def timeFuture(fn: => Future[_]) = {
  val t0 = System.currentTimeMillis
  Await.result(fn, Inf)
  println((System.currentTimeMillis - t0) / 1000.0 + "s")
}

以下两个print~2.5s:

// Use Future.reduce directly (Future.traverse is no different)
timeFuture { Future.reduce(futures)(slowAdd) }

// First wait for all results to come in, convert to Future[List], …

我使用嵌入式(C)编写了一个使用8051微控制器的腕表程序.总共有6个7段显示器:

         _______________________
        |      |       |        |   two 7-segments for showing HOURS
        | HR   | MIN   |   SEC  |   two 7-segments for showing MINUTES and
        |______._______.________|   two 7-segments for showing SECONDS
          7-segment LED display

为了更新小时,分钟和秒,我们使用了3个for循环.这意味着首先会更新秒数,然后是分钟数,然后是小时数.然后我问我的教授为什么我们不能同时更新(我的意思是一小时后增加小时数而不等待更新的会议记录).他告诉我,由于顺序执行指令,我们无法进行并行处理.

题:

一张数字生日贺卡,可同时连续播放音乐,同时闪烁LED.数字闹钟会在特定时间发出哔哔声.虽然它正在产生声音,但时间将继续更新.所以声音和时间增量都是并行运行的.他们是如何通过顺序执行实现这些结果的？

如何在微控制器中同时运行多个任务(调度)？

我一直在寻找并发和并行之间的区别.我在Rob上发表了一篇关于并发和并行之间差异的话题.他的演讲是在Google Go语言的背景下进行的.根据我的理解,并发性是能够处理多个事物的设计,但并行性是同时多个事物的物理执行.

根据我的理解,Node.js作为单个线程进程运行.因此,尽管Node通过实现回调等方式支持并发,它是否能够处理任务的并行执行？它可以设置为在不同的处理器上运行多个线程吗？