问题标签 [concurrent-processing]

我有一个场景，我需要同时将多个 PDF 文件转换为 PNG 图像。虽然它可以使用 Java 8 Parallel Streams 来实现，但我需要它在 Java 7 上执行。

这是我用来在没有并行处理的情况下使用 PDFBox 将 PDF 转换为 PNG 的片段。

请提出一些解决此问题的方法。

我希望根据某些字段值确保并行流中的顺序。也许我会介绍一些抽象的例子：

假设我们有类 User

我们有这些用户的列表：

我想强制并行流按优先级处理每个用户，假设我们有 100 个优先级为 0 的用户、100 个优先级为 1 的用户和 100 个优先级为 2 的用户。

我只想在优先级 1 完成和优先级 0 完成时启动具有优先级 2 的进程用户。

我认为

mvn 安装 -T 4

可能是我正在寻找的方法（第一个构建独立模块）。可以在java流中做到这一点吗？也可以使用替代品。

我的方法是按优先级划分到特定列表，然后按列表处理列表

我需要对一些大数据帧的不同切片进行一些计算。

假设我有 3 个大数据框df1，df2并且df3.
每个都有一"Date"列。
我需要根据日期切片对这些数据帧进行一些计算，并且由于每次迭代都独立于其他迭代，因此我需要同时进行这些迭代。

所以我定义了我想要的函数，在每个子进程中，首先在进程中创建一个 , ,的切片df1，然后进行其他计算。df2df3

由于df1和是全局数据框，df2因此df3我需要将它们指定为函数中的参数。否则将无法识别。

如下所示：

并发处理配置如下：

然而，似乎只有一个核心在执行时上升。

我想由于每个子进程都想访问全局数据帧df1，df2并且df3，应该有一个子进程的共享内存，当我通过网络搜索时，我想我必须使用multiprocessing.manager()，但我不确定如何使用它或如果我使用它是对的？

我实际上是并发处理概念的新手，如果有人可以提供帮助，我将不胜感激。

PS：看来我的问题与这篇文章类似。但是，它没有一个公认的答案。

I wanna try something like download from a collection of image in a minimum of 3 concurrent download but, as soon as one of the 3 finished downloading add a new download from the list or must wait to add a new download until a download is finished from 3 concurrent download. How do I implement something like that?

So far I have tried this, but it seems to download all without waiting to be finished from at least 1 from the 3 concurrent download.

我正在使用 zerodha 经纪人的 API 进行 Algo 交易项目。我正在尝试进行多线程处理，以节省调用 API 函数以一次获取 50 只股票的历史数据的昂贵操作，然后将我的策略应用于买入/卖出。

这是我的代码：

1. 历史数据功能：

1. 策略功能：

3. 并发调用上述函数：

在在线尝试了许多解决方案两天后，现在有 2 个问题我无法解决。

1. API 服务器存在一个瓶颈，它每秒接受 20 个 API 调用，如果调用更多则拒绝。列表中的股票总数为 50，我想做的是一次获取 20 只股票的数据，然后在下一次获取另外 20 只，然后在第三轮中剩下 10 只。总股票清单很快就会有 200 只股票，这就是为什么连续执行对于我的策略来说太昂贵了。
2. 同时运行此函数时，一次创建的线程太多，API 请求超过......并且print('time is : ', end-start)在我运行第三个单元格时立即运行。

那么如何在所有线程完成执行之前阻止代码离开内部 for 循环。

和

我每秒最多获得 20 个线程的方法是否正确？我应该在sleep(1)某个地方放置一个吗？

我正在使用 python 的multiprocessing模块同时处理多个函数。每个衍生进程的函数都获取一些初始输入参数，以及一个Pipe将其结果发回的连接。由于各种原因，我必须使用像这样的单个进程，即Pool.map_async()-methods 之类的工具不可用。

有时，我需要终止一个需要很长时间才能完成的进程。

根据流程文档：

警告：如果在相关进程正在使用管道或队列时使用此方法，则管道或队列可能会损坏并且可能无法被其他进程使用。类似地，如果进程获得了锁或信号量等，那么终止它很可能导致其他进程死锁。

我不担心第一部分，因为每个进程都有自己的管道对象，但是我如何确定一个进程是否“获得了锁或信号量”，和/或以对我的其余部分安全的方式终止程序？

我正在使用一个函数在 mongo 和 oracle 之间迁移数据。此函数将 3 个参数作为输入：与 mongo 的连接、与 oracle 的连接以及我在 mongo 中用作键的 mongo 数据库的实体。我正在遍历一个实体列表，所以代码看起来像这样：

有没有办法并行化这个 for 循环？我不可能使用 python3 来完成这项任务。

我正在使用该multiprocessing模块，但我无法使用该模块打印函数的任何输出。

我什至尝试stdout使用该sys模块刷新。

问题

我们的PROCESSING SERVICE为 UI、API 和内部客户端提供服务，并监听来自Kafka的命令。少数 API 客户端可能会在短时间内创建大量生成任务（一个任务是 N 条消息）。使用 Kafka，我们无法控制命令分发，因为每个命令都到达由一个处理实例（又名工作人员）使用的分区。因此，在处理 API 请求时，UI 请求可能会等待太久。

在理想的实现中，我们应该均匀地处理所有任务，无论其大小。处理服务的容量分布在所有活动任务中。而且即使集群负载很重，我们也始终明白，已经到达的新任务几乎可以立即开始处理，至少在所有其他任务的处理结束之前。

解决方案

相反，我们想要一个看起来更像下图的架构，其中每个客户和端点的组合都有单独的队列。这种架构为我们提供了更好的隔离，以及基于每个客户动态调整吞吐量的能力。 在制作人的一边

• 任务来自客户端
• 立即为此任务创建队列
• 将所有消息发送到此队列

站在消费者一边

• 在一个过程中，您不断更新队列列表
• 在其他进程中，您遵循此列表并使用每个队列中的例如 1 条消息
• 规模消费者

问题

这样的问题有什么通用的解决方案吗？使用 RabbitMQ 或任何其他工具。Н从历史上看，我们在项目中使用 Kafka，所以如果有任何方法使用 - 这太棒了，但我们可以使用任何技术来解决问题。

我正在尝试使用 yahoo Finance 将 1700 个（公司）股票市场数据集下载到 csv 文件中，并且它成功地存储了它们，我使用了 while 循环，我运行，while 循环直到 1700 次，几乎花了 2 多小时，我可以在python中使用并行编程来节省时间吗？

我还对数据执行了一些不同的操作，并在 while 循环中将其存储在 MySQL 数据库中。请帮我解决这个问题