java - 阻塞队列、单个生产者-多个消费者同步问题

Question

我想在将代码投入生产之前进行确认。

我正在使用 LinkedBlockingQueue，如下所示。 我有一个生产者线程和五个消费者线程。 Consumer线程进行DB操作，所以我将其增加到5个以加快进程。

现在，有什么办法可以在代码或数据库级别遇到(同步/行的多次更新/或任何我需要注意的问题)问题。

下面是我的完整代码。

private int numberOfConsumers = 5;
private LinkedBlockingQueue<RequestParameters> processorQueue;


public void init(){
        processorQueue = new LinkedBlockingQueue<RequestParameters>();

        for (int i = 0; i < numberOfConsumers ; i++) {
            try {
                QueueConsumer c = new QueueConsumer(processorQueue);
                c.setName("ThreadName-"+i);
                c.start();

            } catch (Exception e) {
                logger.error("", e);
            }
        }
        this.postCallProcessorDaemon = this;
    }



    class QueueConsumer extends Thread {
        private  Logger log = Logger.getLogger(QueueConsumer.class);
        private final BlockingQueue<RequestParameters> queue;
        QueueConsumer(BlockingQueue<RequestParameters> q) { queue = q; }

      public void run() {   
            try {
                while (true) {
                    RequestParameters rp = queue.take();
                    consumeRecord(rp);
                }
            } catch (Exception ex) {
                log.error("Exception in run method", ex);
            }
        }


    void consumeRecord(RequestParameters requestParameters) 
    {
    try{
           process(requestParameters);
        } catch (Throwable e) {
            log.error("Exception",e);
        }
    }


    private void process(RequestParameters requestParameters) throws Exception{
        for (Action each : allActions) {
            if(each.doProcessing(requestParameters)){
                boolean status = each.process(requestParameters);
            }
        }
    }

}


public boolean process(RequestParameters parameters) {
   //In my process method, I am inserting rows in table based on data in Queue.
}

我用于 ORM 映射的 HBM 文件

<class name="CampaignSubscriptionsSummary" table="campaign_subscriptions_summary">
    <composite-id name="id" class="com.on.robd.config.db.reports.CampaignSubscriptionsSummaryId">
        <key-property name="reportTime" type="timestamp">
            <column name="REPORT_TIME" length="19" />
        </key-property>
        <key-property name="campaignId" type="long">
            <column name="CAMPAIGN_ID" />
        </key-property>
        <key-property name="recipient" type="string">
            <column name="RECIPIENT" length="24" />
        </key-property>
        <key-property name="selectedPack" type="string">
            <column name="SELECTED_PACK" length="256" />
        </key-property>
    </composite-id>
    <many-to-one name="campaign" class="com.on.robd.config.db.campaigns.Campaign" update="false" insert="false" fetch="select">
        <column name="CAMPAIGN_ID" not-null="true" />
    </many-to-one>
    <many-to-one name="campaignSubscriptionsStatusDim" class="com.on.robd.config.db.reports.CampaignSubscriptionStatusDim" fetch="select">
        <column name="SUBSCRIPTION_STATUS_ID" not-null="true" />
    </many-to-one>
    <property name="sender" type="string">
        <column name="SENDER" length="24" not-null="true" />
    </property>
    <property name="callDuration" type="java.lang.Integer">
        <column name="CALL_DURATION" />
    </property>
    <property name="dtmfInput" type="string">
        <column name="DTMF_INPUT" length="16" not-null="true" />
    </property>
    <property name="promptForPack" type="string">
        <column name="PROMPT_FOR_PACK" length="256" />
    </property>
    <property name="wavFile" type="string">
        <column name="WAV_FILE" length="256" />
    </property>
    <property name="subscriptionUrl" type="string">
        <column name="SUBSCRIPTION_URL" length="256" />
    </property>
    <property name="language" type="string">
        <column name="LANGUAGE" length="256" />
    </property>
    <property name="reobdTime" type="timestamp">
            <column name="REOBD_TIME" length="19" />
    </property>
    <many-to-one name="campaignTypeDim" class="com.on.robd.config.db.reports.CampaignTypeDim" fetch="select">
        <column name="CAMPAIGN_TYPE_ID" not-null="true" />
    </many-to-one>
</class>

Answer 1

很可能您会在使用者之间遇到数据库锁定问题，但这取决于您的数据库供应商。检查您的数据库供应商文档，看看它是否在写入时使用表、行或分区锁定。

但无论如何，如果使用批量插入，您将获得更好的写入性能。请参阅http://docs.jboss.org/hibernate/orm/3.3/reference/en-US/html/batch.html 。最简单的批处理方法是让您的生产者将一组 RequestParameters 推送到您的 BlockingQueue 中。这还可以减少与数据库的打开连接数量，这也有助于提高性能。