forked from Archives/Athou_commafeed
use guava for lock handling
This commit is contained in:
15
pom.xml
15
pom.xml
@@ -31,16 +31,6 @@
|
|||||||
<enabled>true</enabled>
|
<enabled>true</enabled>
|
||||||
</snapshots>
|
</snapshots>
|
||||||
</repository>
|
</repository>
|
||||||
<repository>
|
|
||||||
<id>jklm.releases</id>
|
|
||||||
<url>http://mvn.jkeylockmanager.de</url>
|
|
||||||
<releases>
|
|
||||||
<enabled>true</enabled>
|
|
||||||
</releases>
|
|
||||||
<snapshots>
|
|
||||||
<enabled>false</enabled>
|
|
||||||
</snapshots>
|
|
||||||
</repository>
|
|
||||||
</repositories>
|
</repositories>
|
||||||
|
|
||||||
<build>
|
<build>
|
||||||
@@ -244,11 +234,6 @@
|
|||||||
<artifactId>joda-time</artifactId>
|
<artifactId>joda-time</artifactId>
|
||||||
<version>2.2</version>
|
<version>2.2</version>
|
||||||
</dependency>
|
</dependency>
|
||||||
<dependency>
|
|
||||||
<groupId>de.jkeylockmanager</groupId>
|
|
||||||
<artifactId>jkeylockmanager</artifactId>
|
|
||||||
<version>1.0.0</version>
|
|
||||||
</dependency>
|
|
||||||
|
|
||||||
<dependency>
|
<dependency>
|
||||||
<groupId>net.java.dev.rome</groupId>
|
<groupId>net.java.dev.rome</groupId>
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ import java.util.concurrent.ArrayBlockingQueue;
|
|||||||
import java.util.concurrent.RejectedExecutionHandler;
|
import java.util.concurrent.RejectedExecutionHandler;
|
||||||
import java.util.concurrent.ThreadPoolExecutor;
|
import java.util.concurrent.ThreadPoolExecutor;
|
||||||
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.TimeUnit;
|
||||||
|
import java.util.concurrent.locks.Lock;
|
||||||
|
|
||||||
import javax.annotation.PostConstruct;
|
import javax.annotation.PostConstruct;
|
||||||
import javax.annotation.PreDestroy;
|
import javax.annotation.PreDestroy;
|
||||||
@@ -26,10 +27,7 @@ import com.commafeed.backend.model.FeedSubscription;
|
|||||||
import com.commafeed.backend.pubsubhubbub.SubscriptionHandler;
|
import com.commafeed.backend.pubsubhubbub.SubscriptionHandler;
|
||||||
import com.commafeed.backend.services.ApplicationSettingsService;
|
import com.commafeed.backend.services.ApplicationSettingsService;
|
||||||
import com.commafeed.backend.services.FeedUpdateService;
|
import com.commafeed.backend.services.FeedUpdateService;
|
||||||
|
import com.google.common.util.concurrent.Striped;
|
||||||
import de.jkeylockmanager.manager.KeyLockManager;
|
|
||||||
import de.jkeylockmanager.manager.KeyLockManagers;
|
|
||||||
import de.jkeylockmanager.manager.LockCallback;
|
|
||||||
|
|
||||||
@Singleton
|
@Singleton
|
||||||
public class FeedRefreshUpdater {
|
public class FeedRefreshUpdater {
|
||||||
@@ -37,8 +35,6 @@ public class FeedRefreshUpdater {
|
|||||||
protected static Logger log = LoggerFactory
|
protected static Logger log = LoggerFactory
|
||||||
.getLogger(FeedRefreshUpdater.class);
|
.getLogger(FeedRefreshUpdater.class);
|
||||||
|
|
||||||
private static final KeyLockManager lockManager = KeyLockManagers.newLock();
|
|
||||||
|
|
||||||
@Inject
|
@Inject
|
||||||
FeedUpdateService feedUpdateService;
|
FeedUpdateService feedUpdateService;
|
||||||
|
|
||||||
@@ -53,7 +49,7 @@ public class FeedRefreshUpdater {
|
|||||||
|
|
||||||
@Inject
|
@Inject
|
||||||
ApplicationSettingsService applicationSettingsService;
|
ApplicationSettingsService applicationSettingsService;
|
||||||
|
|
||||||
@Inject
|
@Inject
|
||||||
MetricsBean metricsBean;
|
MetricsBean metricsBean;
|
||||||
|
|
||||||
@@ -61,12 +57,14 @@ public class FeedRefreshUpdater {
|
|||||||
FeedSubscriptionDAO feedSubscriptionDAO;
|
FeedSubscriptionDAO feedSubscriptionDAO;
|
||||||
|
|
||||||
private ThreadPoolExecutor pool;
|
private ThreadPoolExecutor pool;
|
||||||
|
private Striped<Lock> locks;
|
||||||
|
|
||||||
@PostConstruct
|
@PostConstruct
|
||||||
public void init() {
|
public void init() {
|
||||||
ApplicationSettings settings = applicationSettingsService.get();
|
ApplicationSettings settings = applicationSettingsService.get();
|
||||||
int threads = Math.max(settings.getDatabaseUpdateThreads(), 1);
|
int threads = Math.max(settings.getDatabaseUpdateThreads(), 1);
|
||||||
log.info("Creating database pool with {} threads", threads);
|
log.info("Creating database pool with {} threads", threads);
|
||||||
|
locks = Striped.lock(threads);
|
||||||
pool = new ThreadPoolExecutor(threads, threads, 0,
|
pool = new ThreadPoolExecutor(threads, threads, 0,
|
||||||
TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(
|
TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(
|
||||||
100 * threads));
|
100 * threads));
|
||||||
@@ -130,12 +128,13 @@ public class FeedRefreshUpdater {
|
|||||||
|
|
||||||
private void updateEntry(final Feed feed, final FeedEntry entry,
|
private void updateEntry(final Feed feed, final FeedEntry entry,
|
||||||
final List<FeedSubscription> subscriptions) {
|
final List<FeedSubscription> subscriptions) {
|
||||||
lockManager.executeLocked(entry.getGuid(), new LockCallback() {
|
Lock lock = locks.get(entry.getGuid());
|
||||||
@Override
|
lock.lock();
|
||||||
public void doInLock() throws Exception {
|
try {
|
||||||
feedUpdateService.updateEntry(feed, entry, subscriptions);
|
feedUpdateService.updateEntry(feed, entry, subscriptions);
|
||||||
}
|
} finally {
|
||||||
});
|
lock.unlock();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private void handlePubSub(final Feed feed) {
|
private void handlePubSub(final Feed feed) {
|
||||||
|
|||||||
Reference in New Issue
Block a user