我有一个架构问题,关于如何在Java/Java EE中处理事务性和可伸缩性的大任务.

一般的挑战

我有一个Web应用程序(Tomcat现在,但不应该限制解决方案空间,所以只需要用它来说明我想要实现的目标).这个Web应用程序分布在几个(虚拟和物理)节点上,连接到中央DBMS(在这种情况下是MySQL,但同样,这不应该限制解决方案......)并且能够处理大约1000个用户,服务页面,正如你对平均基于网络的信息系统所期望的那样.

现在,有一些任务影响了大部分数据,系统应该进行优化,以便合理地快速执行这些任务.(比顺序处理所有内容更快,即).所以我将任务并行并将其分布在几个(或所有)节点上:

(注意:处理的数据部分是独立的,因此这里没有数据库或锁定冲突).

问题是,我希望(整个)任务是事务性的.因此,如果其中一个并行子任务失败,我希望将所有其他任务作为结果回滚.否则,从域的角度来看,系统将处于可能不一致的状态.

目前的实施

正如我所说,当前的实现使用Tomcat和MySQL.节点使用JMS进行通信(因此有一个JMS服务器,调度程序为每个子任务发送消息;执行程序从消息队列中获取任务,执行它们,并将结果发布到调度程序收集的结果队列中.调度程序阻塞并等待所有结果进入,如果一切正常,它将以OK状态终止.

这里的问题是所有执行程序都有自己的本地事务上下文,因此图片看起来像这样:

如果由于某种原因,其中一个子任务失败,则回滚本地事务并且调度程序获得错误结果.(这里有一些故障保护机制,它试图重复失败的事务,但我们假设由于某种原因,一个任务无法完成).问题是系统现在处于一个状态,其中除了一个之外的所有事务都已提交并完成.而且因为我不能让最后一笔交易成功完成,所以我无法摆脱这种状态.

可能的解决方案

这些是我到目前为止所遵循的想法:

• 我自己可以以某种方式实现特定于域的回滚机制.因为分发器知道已经执行了哪些任务,所以它可以显式地恢复效果(例如,在某处存储旧值并将已提交的值恢复回先前的值).当然,在这种情况下,我必须保证其他进程之间不会发生任何变化,所以只要大型操作正在运行,我还必须将系统设置为只读状态.或多或少,我需要模拟业务逻辑中的事务...

• 我可以选择不在一个大事务中并行化并在单个节点上执行所有操作(但正如开头所述,我需要加快处理速度,因此这不是一个选项......)

• 我试图了解一般的XATransactions或分布式事务,但这似乎是一个高级的Java EE功能,它没有在所有Java EE服务器中实现,并且不能真正解决这个基本问题,因为似乎没有是一种在异步调用中将事务上下文传递到远程节点的方法.(例如,EJB规范3.1的 4.5.3部分:"客户端事务上下文不会通过异步方法调用进行传播.从Bean Developer的视图来看,永远不会有来自客户端的事务上下文.")

问题

我忽略了什么吗？是不是可以在多个节点上异步分发任务,同时具有可以作为整体回滚的(共享)事务状态？

感谢任何指示,提示,命题......

为什么这不符合我的想法:

benjamin@benjamin-VirtualBox:~$ julia -p 3
julia> @everywhere(function foom(bar::Vector{Any}, k::Integer) println(repeat(bar[2],bar[1])); return bar; end)
julia> foo={{1,"a"},{2,"b"},{3,"c"}}
julia> pmap(foom, foo, 5)
    From worker 2:  a
1-element Array{Any,1}:
 {1,"a"}

这就是它的全部输出.我期待pmap迭代foo中的每个元组并在其上调用foom.

编辑:

当我没有传递其他参数时,它可以正常工作:

julia> @everywhere(function foom(bar::Vector{Any}) println(repeat(bar[2],bar[1])); return bar; end)

julia> pmap(foom, foo)
    From worker 3:  bb
    From worker 2:  a
    From worker 4:  ccc
3-element Array{Any,1}:
 {1,"a"}
 {2,"b"}
 {3,"c"}

如何将更多参数传递给pmap？

我正在12节点集群上运行并行进程.

并且想知道在foreach调用期间是否有办法获取node-id或node-number或node-name ？

像这样的东西:

foreach(i = 1:12, .combine=c) %dopar% {node.name()}

这将有助于处理文件.

我正在尝试处理> 10000 xts保存在磁盘上的对象,每个加载到R时大约为0.2 GB.我想使用foreach并行处理这些对象.我的代码适用于100 xts对象,我在内存中预先加载,导出等.但是在> 100 xts对象后,我在我的机器上达到了内存限制.

我想要做的例子:

require(TTR)
require(doMPI)
require(foreach)

test.data <- runif(n=250*10*60*24)

xts.1 <- xts(test.data, order.by=as.Date(1:length(test.data)))
xts.1 <- cbind(xts.1, xts.1, xts.1, xts.1, xts.1, xts.1)

colnames(xts.1) <- c("Open", "High", "Low", "Close", "Volume", "Adjusted")

print(object.size(xts.1), units="Gb")

xts.2 <- xts.1
xts.3 <- xts.1
xts.4 <- xts.1

save(xts.1, file="xts.1.rda")
save(xts.2, file="xts.2.rda")
save(xts.3, file="xts.3.rda")
save(xts.4, file="xts.4.rda")

names <- c("xts.1", "xts.2", "xts.3", "xts.4")

rm(xts.1)
rm(xts.2)
rm(xts.3)
rm(xts.4)

cl <- startMPIcluster(count=2) # Use 2 cores
registerDoMPI(cl)

result <- foreach(name=names, 
                  .combine=cbind, 
                  .multicombine=TRUE, 
                  .inorder=FALSE, 
                  .packages=c("TTR")) %dopar% { …

我一直在敲打这个问题好几个小时,我总是因为线程争用而吃掉了并行化循环的性能改进.

我正在尝试计算8位灰度千兆像素图像的直方图.读过"CUDA by example"一书的人可能会知道它的来源(第9章).

该方法非常简单(导致非常紧凑的循环).它基本上就是这样

    private static void CalculateHistogram(uint[] histo, byte[] buffer) 
    {
        foreach (byte thisByte in buffer) 
        {
            // increment the histogram at the position
            // of the current array value
            histo[thisByte]++;
        }
    }

其中buffer是1024 ^ 3个元素的数组.

在最新的Sandy Bridge-EX CPU构建中,10亿个元素的直方图在一个核心上运行1秒钟.

无论如何,我尝试通过在所有核心之间分配循环来加速计算,最终得到的解决方案慢了50倍.

    private static void CalculateHistrogramParallel(byte[] buffer, ref int[] histo) 
    {
        // create a variable holding a reference to the histogram array
        int[] histocopy = histo;

        var parallelOptions = new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount };

        // loop through the …

我有一个非常大的 data.frame,我想应用一个相当复杂的函数,计算一个新列.我想要并行完成.这类似于r listserve上发布的问题,但第一个答案是错误的,第二个答案是无益的.

由于parallel包装的原因,我已经弄明白了,除了如何将输出重新放回数据框.这是一个显示我所得到的MWE:

library(parallel)

# Example Data
data <- data.frame(a = rnorm(200), b = rnorm(200),  
                   group = sample(letters, 200, replace = TRUE))

# Break into list
datagroup <- split(data, factor(data$group))

# execute on each element in parallel
options(mc.cores = detectCores())
output <- mclapply(datagroup, function(x) x$a*x$b)

结果output是一个数字向量列表.我需要将它们添加到我可以添加的列中data.我一直在寻找do.call(cbind, ...),但我有两个名称相同的列表,而不是我加入的单个列表.melt(output)给我一个向量,但它的行与...的顺序不同data.

如果我认为问题是并行化的候选者,例如矩阵乘法或其他一些问题而且我使用的是英特尔i7 haswell双核,我是否可以通过某种方式将并行执行与同一程序的顺序版本进行比较,或者将matlab优化为程序到我的架构(dualcore,quadcore ..)？我想知道从一个好的基准平行程序添加更多处理器的速度.

例如,对于任何给定的集合Map,

val in = Array( Map("a" -> 1,  "b" -> 2),
                Map("a" -> 11, "c" -> 4),
                Map("b" -> 7,  "c" -> 10))

如何使用aggregate上in.par,从而地图合并成

Map ( "a" -> 12, "b" -> 9, "c" -> 14 )

注意Map合并已被多次询问,但寻找aggregate并行集合的解决方案.

非常感谢

在clearmake中,有一个选项可以传递主机名,以便在这些主机上运行多个作业但在gmake中没有选项可以传递多个主机,尽管可以传递多个作业.我想知道我如何模仿这种clearmake功能？

在Windows 7上使用R 2.14.1

使用R中的并行程序包,我正在尝试利用我的网络上可用的本地计算机之外的核心,我连接的所有远程主机都是相同的Windows计算机.

命令的基本形式就是建立连接.

library(parallel)
#assume 8 cores per machine
cl<-makePSOCKcluster(c(rep("localhost", 8), rep("otherhost", 8)))

当然,尝试调试这些东西可能相当棘手,但这就是我所处的位置.

如果我指定manual = TRUE标志如下

cl<-makePSOCKcluster(c(rep("localhost", 8), rep("otherhost", 8)), manual=TRUE)

连接到远程主机并运行并行进程没有问题.计算机与我正在使用的计算机具有相同的设置.但是,如果未设置此手动标志,则连接命令将挂起.

这似乎向我表明,由于手动标志绕过ssh以建立与主机的连接,因此当手动= FALSE时,ssh是问题.

目前无法保证远程计算机上有ssh.问题是,鉴于我拥有远程主机的所有相关Windows登录信息,并且我无法更改远程计算机上的设置,我将如何连接到远程计算机上的核心,并且在R中并行指定包,而不指定手动=真正？

或者,如果必须安装ssh才能实现此目的,让我们假设所有计算机都有ssh.如何在不绕过ssh的情况下连接到远程计算机上的核心？

如果您需要更多信息,请告诉我,我很感激时间.

更新1

14年8月26日

感谢Steve Weston的见解.我将提供一个更新,其中包含我使用的确切工具和设置,以便在系统启动并运行时使其正常工作.

如果您还有其他任何可以添加的内容,可以通过makePSOCKcluster从Windows机器远程连接到Windows机器的最佳路径,其中手动标志设置为FALSE,请随意发表评论或发布.