标签: concurrent-programming

在这里,我再次提出有关多线程和我的并发编程类的练习的问题.

我有一个多线程服务器 - 使用.NET 异步编程模型实现 - 带GET(下载)和PUT(上传)文件服务.这部分已完成并经过测试.

碰巧问题的陈述说这个服务器必须具有对服务器响应时间影响最小的日志记录活动,并且应该由低优先级线程 - 记录器线程支持 - 为此效果创建.所有日志消息都应由生成它们的线程传递给此记录器线程,使用可能不会锁定调用它的线程的通信机制(除了必要的锁定以确保互斥)并假设某些日志消息可能被忽略.

这是我目前的解决方案,请帮助验证这是否是解决所述问题的方法:

using System;
using System.IO;
using System.Threading;

// Multi-threaded Logger
public class Logger {
    // textwriter to use as logging output
    protected readonly TextWriter _output;
    // logger thread
    protected Thread _loggerThread;
    // logger thread wait timeout
    protected int _timeOut = 500; //500ms
    // …

我今天参加了面试,面试官问我以下问题:

重新入侵和互斥是线程安全的吗？你能解释一下原因吗？

我对并发编程比较陌生,无法回答它..但我说......

相互排斥是线程安全的.但重新入门不是,这就是为什么我们有可重入锁定的原因.

面试官转到下一个问题但是到了另一个地方......我想我搞砸了这个......

当他问我这个时,他期待我说什么？

我正在使用异步I/O,因为它不会阻塞调用线程并在幕后执行线程处理.如果我在同一个Stream上调用多个异步操作(如BeginWrite()),我是否需要担心数据缓冲区内容是否混合在一起？

我们假设我想发送3个缓冲区:

Buffer1: 1111111111
Buffer2: 2222222222
Buffer3: 3333333333

我不介意缓冲区是否以错误的顺序发送,所以

333333333311111111112222222222

没问题,但缓冲内容是否可能完全混合在一起？

122213121212122333313111223333

PS:我100%肯定有人已经以某种形式问过这个......

我在模拟系统上工作,在每个时间步,我必须模拟许多模型.我使用FixedThreadPool来加速计算:

ExecutorService executor = Executors.newFixedThreadPool(nThread);
for (Model m : models) {
  executor.execute( m.simulationTask() );
}
executor.shutdown();
while ( ! executor.awaitTermination(10, TimeUnit.MINUTES) ) { 
  System.out.println("wait"); 
}

现在,执行程序execute()在调用后不能用于新任务shutdown().有没有办法重置执行程序,所以我可以在下一个模拟步骤中重用现有的执行程序(及其线程)？

如何从地址窃取2个MSB进行原子操作？我正在努力做一个单词CAS

一个例子

public class Node
{
    long key;
    long value;
    Node lchild; // format is flag1,flag2,address
    Node rchild; // format is flag1,flag2,address
}

public void createNode()
{
    Node n1 = new Node(); //this should create a node with format 0,0,address1
}

public void setFlag1(Node n1)
{
    Now the new address should be in format 1,0,address1
}

public void setFlag2(Node n1)
{
    Now the new address should be in format 0,1,address1
}

AtomicReference如果我只需要一个额外的标志,可以使用. AtomicStampedReference可以使用,但它没有效率,因为它创建了一个包含timeStamp和引用的额外框.

在从指针窃取位时讨论了C中的类似问题

我有以下TCP客户端 - 服务器通信结构:

• 在服务器启动服务器上启动接受器线程,它接受客户端连接并将ServerSocket传递给它.
• 当客户端连接到达时,接受者线程调用ServerSocket上的accept()并将客户端处理作业提交给工作线程(通过执行器/线程池)并向其提供客户端套接字.
• 循环中的Worker从客户端套接字流中读取数据,处理它并发送回复.

问题是如何优雅地停止整个系统？我可以通过关闭ServerSocket来停止接受器线程.它会导致accept()阻塞调用抛出SocketException.但是如何阻止工人呢？他们从流中读取并且此调用正在阻止.根据这个流不会抛出InterruptedException,因此worker不能被中断().

看起来我需要从另一个线程关闭工作套接字,对吗？为此,套接字应该是一个公共字段,或者应该在worker中提供一个方法来关闭它.这会不错？或者可能是我的整个设计有缺陷？

我已经阅读了博客,但我不确定他的结论是否正确:

http://www.javacodegeeks.com/2010/09/java-best-practices-queue-battle-and.html#ixzz1seaiSLwp

他说:从提供的性能结果可以看出,LinkedBlockingQueue实现了最佳的组合(添加和删除元素)性能结果,应该是实现生产者 - 消费者schenarios的头号候选者.

我想知道,如果我不在我的代码中使用锁定,那么它会不会更快？

那么为什么LinkedBlockingQueue比无锁队列(ConcurrentLinkedQueue)更快？

谢谢 !

我有一个情况,我正在进行几个(比如四个)ajax调用(使用AngularJS http get,如果这很重要)我希望每次调用回调并增加一个计数器,所以我可以知道所有(四个)线程何时有完成.

我担心的是,由于JavaScript没有任何可与Java的"synchronized"或"volatile"关键字相媲美的东西,因此多个并发线程可能会在递增计数器时发生冲突,从而错过了一些增量.

换句话说,两个线程同时出现,并且都读取计数器,获得相同的值(例如100).然后两个线程都递增该计数器(到101)并存储新值,不料,我们错过了一个计数(101而不是102)!

我知道JavaScript应该是单线程的,但也有例外.

这种情况是否可以在浏览器中使用,如果是这样,有没有办法防范它？

我试图使用channels/STM在Haskell中实现消息传递.也许这是一个糟糕的想法,并且有一种更好的方法来实现/使用Haskell中的消息传递.如果是这种情况,请告诉我; 然而,我的任务已经打开了关于并发Haskell的一些基本问题.

我听说过STM很棒的东西,尤其是Haskell中的实现.既然它支持阅读和写作,并且有一些安全上的好处,我认为可以从那里开始.这提出了我最大的问题:是吗

msg <- atomically $ readTChan chan

其中chan是一个TChan Int,导致等待通道等待它的等待？

考虑以下程序:

p chan = do
    atomically $ writeTChan chan 1
    atomically $ writeTChan chan 2

q chan = do
    msg1 <- atomically $ readTChan chan 
    msg2 <- atomically $ readTChan chan
    -- for testing purposes
    putStrLn $ show msg1 
    putStrLn $ show msg2

main = do
    chan <- atomically $ newTChan
    p chan
    q chan

用ghc --make -threaded编译它,然后运行程序,实际上你得到1然后打印到控制台2.现在,假设我们这样做

main = do 
    chan <- atomically $ newTChan
    forkIO $ p …

我有一个4台机器的集群，1台master，3台worker，每台128G内存，64个核心。我在独立模式下使用 Spark 1.5.0。我的程序使用 JDBC 从 Oracle 表中读取数据，然后执行 ETL、操作数据，并执行 k-means 等机器学习任务。

我有一个 DataFrame (myDF.cache())，它是与其他两个 DataFrame 的连接结果，并进行缓存。DataFrame包含2700万行，数据大小约为1.5G。我需要过滤数据并计算24直方图，如下：

val h1 = myDF.filter("pmod(idx, 24) = 0").select("col1").histogram(arrBucket) 
val h2 = myDF.filter("pmod(idx, 24) = 1").select("col1").histogram(arrBucket) 
// ...... 
val h24 = myDF.filter("pmod(idx, 24) = 23").select("col1").histogram(arrBucket) 

问题：

1. 由于我的 DataFrame 已被缓存，因此我希望过滤器、选择和直方图非常快。但每次计算的实际时间约为7秒，这是不可接受的。从 UI 来看，GC 时间需要 5 秒，任务反序列化时间需要 4 秒。我尝试了不同的 JVM 参数，但无法进一步改进。现在我正在使用

   -Xms25G -Xmx25G -XX:MaxPermSize=512m -XX:+UseG1GC -XX:MaxGCPauseMillis=200 \
   -XX:ParallelGCThreads=32 \
   -XX:ConcGCThreads=8 -XX:InitiatingHeapOccupancyPercent=70 
   

   Run Code Online (Sandbox Code Playgroud)

让我困惑的是，数据的大小与可用内存相比根本不算什么。为什么每次运行过滤器/选择/直方图时都会启动 GC？有没有办法减少GC时间和任务反序列化时间？

2. 我必须对 h[1-24] 进行并行计算，而不是顺序计算。我尝试了未来，类似：

   -Xms25G -Xmx25G -XX:MaxPermSize=512m -XX:+UseG1GC -XX:MaxGCPauseMillis=200 \
   -XX:ParallelGCThreads=32 \
   -XX:ConcGCThreads=8 -XX:InitiatingHeapOccupancyPercent=70 
   

   Run Code Online (Sandbox Code Playgroud)

问题是，这里的 Future 仅意味着作业几乎同时提交给调度程序，而不是它们最终被调度并同时运行。这里使用 Future …