検索

package timer;
 
public interface AlarmListener {
    public void onAlarm();
}

package timer;
 
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
 
/**
http://www.02.246.ne.jp/~torutk/javahow2/timer.html#doc1_id120
http://d.hatena.ne.jp/AWAWA/20080423/1208955538
http://www.ibm.com/developerworks/jp/java/library/j-5things5.html
 */
public class PeriodicAlarm implements Runnable {
 
    ScheduledFuture future;
    ScheduledExecutorService executorService;
    final Set<AlarmListener> listeners = new HashSet<>();
    final long initialDelay;
    final long period;
    final TimeUnit timeUnit;
 
    public PeriodicAlarm(long initialDelay, long period, TimeUnit timeUnit) {
        this.executorService = Executors.newScheduledThreadPool(1);
        this.initialDelay = initialDelay;
        this.period = period;
        this.timeUnit = timeUnit;
    }
 
    public void start() {
        this.future = this.executorService.scheduleAtFixedRate(this, this.initialDelay, this.period, this.timeUnit);
    }
 
    public void stop() {
        this.future.cancel(true);
    }
 
    public void shutdown() {
        this.stop();
        this.executorService.shutdown();
    }
 
    public void addAlarmListener(AlarmListener l) {
        synchronized (listeners) {
            listeners.add(l);
        }
    }
 
    public void removeAlarmListener(AlarmListener l) {
        synchronized (listeners) {
            listeners.remove(l);
        }
    }
 
    @Override
    public void run() {
        synchronized (listeners) {
            for (AlarmListener l : this.listeners) {
                l.onAlarm();
            }
        }
    }
}

import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.concurrent.TimeUnit;
import static org.testng.Assert.*;
import org.testng.annotations.Test;
import timer.AlarmListener;
import timer.PeriodicAlarm;
 
public class PeriodicAlarmTest {
 
    public PeriodicAlarmTest() {
    }
 
    class Countup implements AlarmListener {
 
        String name;
        int count = 0;
        final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-DD HH:mm:ss.SSS");
 
        public Countup(String name) {
            this.name = name;
        }
 
        @Override
        public void onAlarm() {
            Calendar c = Calendar.getInstance();
            String snow;
            synchronized (this.sdf) {
                snow = sdf.format(c.getTime());
            }
            count++;
            System.out.println("[" + this.name + "]:now=" + snow + ", count=" + count);
 
        }
    }
 
    @Test
    public void hello() throws InterruptedException {
        PeriodicAlarm a = new PeriodicAlarm(0, 1, TimeUnit.SECONDS);
        a.start();
        Countup c1 = new Countup("c1");
        a.addAlarmListener(c1);
        Thread.sleep(2 * 1000);
        Countup c2 = new Countup("c2");
        a.addAlarmListener(c2);
        Thread.sleep(2 * 1000);
        Countup c3 = new Countup("c3");
        a.addAlarmListener(c3);
        Thread.sleep(2 * 1000);
        a.shutdown();
        assertEquals(c1.count, 6);
        assertEquals(c2.count, 4);
        assertEquals(c3.count, 2);
    }
 
}

package hoge;
 
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
import org.testng.annotations.Test;
 
/**
 *
 * @author FengJing
 */
public class HogeTest {
 
    public HogeTest() {
    }
 
    class JobMailBox<T> extends LinkedBlockingQueue<T> {
 
        protected boolean isShutdown = false;
 
        public void shutdown() {
            this.isShutdown = true;
        }
 
        @Override
        public boolean offer(T e) {
            if (this.isShutdown) {
                return false;
            }
            return super.offer(e);
        }
 
        @Override
        public boolean offer(T e, long timeout, TimeUnit unit) throws InterruptedException {
            if (this.isShutdown) {
                return false;
            }
            return super.offer(e, timeout, unit);
        }
 
        @Override
        public void put(T e) throws InterruptedException {
            if (this.isShutdown) {
                return;
            }
            super.put(e);
        }
 
        @Override
        public T poll(long timeout, TimeUnit unit) throws InterruptedException {
            return super.poll(timeout, unit);
        }
 
        @Override
        public T take() throws InterruptedException {
            return super.take();
        }
 
    }
 
    enum JobEvent {
 
        STARTED, RUNNING, CANCELLED, TERMINATED;
    }
 
    class JobSignal {
 
        public final JobEvent event;
        public final String key;
        public final String message;
 
        public JobSignal(JobEvent event, String key, String message) {
            this.event = event;
            this.key = key;
            this.message = message;
        }
    }
 
    interface JobHandle<R, M> {
 
        public R main(BlockingQueue<M> mailbox) throws Exception;
    }
 
    class JobButler {
 
        protected ExecutorService executorService;
        protected Map<String, Future> jobs = new ConcurrentHashMap<String, Future>();
        protected Map<String, BlockingQueue> mailboxes = new ConcurrentHashMap<String, BlockingQueue>();
        protected final JobMailBox<JobSignal> butlerMailBox = new JobMailBox<>();
 
        public JobButler(ExecutorService executorService) {
            this.executorService = executorService;
        }
 
        public JobMailBox<JobSignal> getButlersMailBox() {
            return this.butlerMailBox;
        }
 
        public <R, M> Future<R> submitNewJob(final String key, final JobHandle<R, M> job) throws InterruptedException {
            final BlockingQueue<M> mailbox = new ArrayBlockingQueue<M>(100);
            Callable<R> c = new Callable<R>() {
                public R call() throws Exception {
                    butlerMailBox.put(new JobSignal(JobEvent.RUNNING, key, "running"));
                    R result = job.main(mailbox);
                    butlerMailBox.put(new JobSignal(JobEvent.TERMINATED, key, "terminated"));
                    return result;
                }
            };
            butlerMailBox.put(new JobSignal(JobEvent.STARTED, key, "started"));
            Future<R> f = this.executorService.submit(c);
            this.jobs.put(key, f);
            this.mailboxes.put(key, mailbox);
            return f;
        }
 
        public Future getFuture(String key) {
            return this.jobs.get(key);
        }
 
        public Future terminateJob(String key) {
            Future f = this.jobs.get(key);
            if (f.isCancelled() || f.isDone()) {
                return f;
            }
            f.cancel(true);
            return f;
        }
 
        public void shutdown() {
            this.executorService.shutdown();
        }
    }
 
    class HogeJob implements JobHandle<String, String> {
 
        private final String name;
 
        public HogeJob(String name) {
            this.name = name;
        }
 
        public String main(BlockingQueue<String> mailbox) throws Exception {
            StringBuilder sb = new StringBuilder(1024);
            sb.append(this.name);
            sb.append(":");
            for (int i = 0; i < 10; i++) {
                sb.append(i);
            }
            return sb.toString();
        }
    }
 
    class CountJob implements JobHandle<Integer, String> {
 
        private int sum = 0;
 
        public Integer main(BlockingQueue<String> mailbox) throws Exception {
            for (int i = 0; i < 10; i++) {
                this.sum += i;
            }
            return new Integer(this.sum);
        }
 
    }
 
    @Test
    public void hoge2() throws InterruptedException, ExecutionException {
        JobButler jb = new JobButler(Executors.newCachedThreadPool());
        Future<String> f1 = jb.submitNewJob("job1", new HogeJob("jon"));
        Future<String> f2 = jb.submitNewJob("job2", new HogeJob("bob"));
        Future<Integer> f3 = jb.submitNewJob("job3", new CountJob());
 
        assertEquals(f1.get(), "jon:0123456789");
        assertEquals(f2.get(), "bob:0123456789");
        assertEquals(f3.get(), new Integer(45));
 
        jb.shutdown();
    }
 
    class SlowCountJob implements JobHandle<Integer, String> {
 
        private int sum = 0;
        private final String name;
 
        public SlowCountJob(String name) {
            this.name = name;
        }
 
        @Override
        public Integer main(BlockingQueue<String> mailbox) throws Exception {
            int i = 0;
            while (!"terminate".equals(mailbox.poll(1000, TimeUnit.MILLISECONDS))) {
                this.sum += i;
                i++;
                System.out.println(this.name + ":" + this.sum);
            }
            return new Integer(this.sum);
        }
 
    }
 
    @Test(expectedExceptions = CancellationException.class)
    public void hoge3() throws InterruptedException, ExecutionException, TimeoutException {
        JobButler jb = new JobButler(Executors.newCachedThreadPool());
        Future<Integer> f1 = jb.submitNewJob("job1", new SlowCountJob("scj-1"));
 
        try {
            Thread.sleep(5000);
        } catch (InterruptedException ignore) {
        }
        // cancel flag was set, but not interrupted ... orphan thread!!!
        f1.cancel(false);
        assertTrue(f1.isCancelled());
        assertTrue(f1.isDone());
        try {
            f1.get(1, TimeUnit.SECONDS);
        } finally {
            jb.shutdown();
        }
    }
 
    @Test(expectedExceptions = CancellationException.class)
    public void hoge4() throws InterruptedException, ExecutionException, TimeoutException {
        JobButler jb = new JobButler(Executors.newCachedThreadPool());
        Future<Integer> f1 = jb.submitNewJob("job1", new SlowCountJob("scj-2"));
 
        try {
            Thread.sleep(5000);
        } catch (InterruptedException ignore) {
        }
        // cancel flag was set, and, inerrupted -> thread terminates.
        f1.cancel(true);
        assertTrue(f1.isCancelled());
        assertTrue(f1.isDone());
        try {
            f1.get(1, TimeUnit.SECONDS);
        } finally {
            jb.shutdown();
        }
    }
 
    class AbnormalEndCountJob implements JobHandle<Integer, String> {
 
        private int sum = 0;
        private final String name;
        private final int stopper;
 
        public AbnormalEndCountJob(String name, int stopper) {
            this.name = name;
            this.stopper = stopper;
        }
 
        @Override
        public Integer main(BlockingQueue<String> mailbox) throws Exception {
            int i = 0;
            while (!"terminate".equals(mailbox.poll(1000, TimeUnit.MILLISECONDS))) {
                this.sum += i;
                i++;
                System.out.println(this.name + ":" + this.sum);
                if (i > stopper) {
                    throw new IllegalStateException("stopper threashold over.");
                }
            }
            return new Integer(this.sum);
        }
 
    }
 
    @Test(expectedExceptions = IllegalStateException.class)
    public void hoge5() throws InterruptedException, ExecutionException, TimeoutException, Throwable {
        JobButler jb = new JobButler(Executors.newCachedThreadPool());
        Future<Integer> f1 = jb.submitNewJob("job1", new AbnormalEndCountJob("abend-job-1", 3));
 
        try {
            Thread.sleep(5000);
        } catch (InterruptedException ignore) {
        }
        // job is already done (stopper = 3 means "After 3 seconds, throw IllegalStateException")
        // so, cancel() does not effect any more. isCancelled() returns false.
        f1.cancel(true);
        assertFalse(f1.isCancelled());
        assertTrue(f1.isDone());
        try {
            f1.get(1, TimeUnit.SECONDS);
        } catch (ExecutionException expected) {
            Throwable t = expected.getCause();
            assertEquals(t.getMessage(), "stopper threashold over.");
            throw t;
        } finally {
            jb.shutdown();
        }
    }
 
    @Test
    public void hoge() throws InterruptedException, ExecutionException {
        ExecutorService es = Executors.newCachedThreadPool();
        Callable<String> c = new Callable<String>() {
            public String call() throws Exception {
                StringBuilder sb = new StringBuilder(1024);
                for (int i = 0; i < 10; i++) {
//                    Thread.sleep(1000);
                    sb.append(i);
                }
                return sb.toString();
            }
        };
        Future<String> f = es.submit(c);
        assertEquals(f.get(), "0123456789");
    }
}

package hoge;
 
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import org.testng.annotations.Test;
 
/**
 *
 * @author FengJing
 */
public class BoheTest {
 
    public BoheTest() {
    }
 
    class JobMailBox<T> extends LinkedBlockingQueue<T> {
 
        protected boolean isShutdown = false;
 
        public void shutdown() {
            this.isShutdown = true;
        }
 
        @Override
        public boolean offer(T e) {
            if (this.isShutdown) {
                return false;
            }
            return super.offer(e);
        }
 
        @Override
        public boolean offer(T e, long timeout, TimeUnit unit) throws InterruptedException {
            if (this.isShutdown) {
                return false;
            }
            return super.offer(e, timeout, unit);
        }
 
        @Override
        public void put(T e) throws InterruptedException {
            if (this.isShutdown) {
                return;
            }
            super.put(e);
        }
 
        @Override
        public T poll(long timeout, TimeUnit unit) throws InterruptedException {
            return super.poll(timeout, unit);
        }
 
        @Override
        public T take() throws InterruptedException {
            return super.take();
        }
 
    }
 
    interface JobHandle<R, M> {
 
        public R main(BlockingQueue<M> mailbox) throws Exception;
    }
 
    class JobTask<V> extends FutureTask<V> {
 
        protected final String key;
 
        protected final JobButler butler;
 
        public JobTask(JobButler butler, String key, Callable<V> callable) {
            super(callable);
            this.key = key;
            this.butler = butler;
        }
 
        public JobTask(JobButler butler, String key, Runnable runnable, V result) {
            super(runnable, result);
            this.key = key;
            this.butler = butler;
        }
 
        @Override
        protected void done() {
            butler.notifyJobTermination(key);
        }
    }
 
    class JobGateway<R, M> {
 
        final Future<R> future;
        final JobMailBox<M> mailbox;
 
        public JobGateway(Future<R> f, JobMailBox<M> mb) {
            this.future = f;
            this.mailbox = mb;
        }
    }
 
    class JobButler {
 
        protected ExecutorService executorService;
        protected Map<String, JobGateway> jobs = new ConcurrentHashMap<>();
 
        public JobButler(ExecutorService executorService) {
            this.executorService = executorService;
        }
 
        public synchronized <R, M> JobGateway<R, M> submitNewJob(final String key, final JobHandle<R, M> job) throws InterruptedException {
            final JobMailBox<M> mailbox = new JobMailBox<>();
            JobTask<R> task = new JobTask<>(this, key, new Callable<R>() {
                @Override
                public R call() throws Exception {
                    R result = job.main(mailbox);
                    return result;
                }
            });
            this.executorService.execute(task);
            JobGateway<R, M> jobgw = new JobGateway(task, mailbox);
            this.jobs.put(key, jobgw);
            return jobgw;
        }
 
        public synchronized JobGateway getJobGateway(String key) {
            return this.jobs.get(key);
        }
 
        synchronized void notifyJobTermination(String key) {
            if (!this.jobs.containsKey(key)) {
                return;
            }
            JobGateway jobgw = this.jobs.get(key);
            jobgw.mailbox.shutdown();
            this.jobs.remove(key);
        }
 
        public void shutdown() {
            this.executorService.shutdown();
        }
    }
 
    class HogeJob implements JobHandle<String, String> {
 
        private final String name;
 
        public HogeJob(String name) {
            this.name = name;
        }
 
        @Override
        public String main(BlockingQueue<String> mailbox) throws Exception {
            StringBuilder sb = new StringBuilder(1024);
            sb.append(this.name);
            sb.append(":");
            for (int i = 0; i < 10; i++) {
                sb.append(i);
            }
            return sb.toString();
        }
    }
 
    class CountJob implements JobHandle<Integer, String> {
 
        private int sum = 0;
 
        @Override
        public Integer main(BlockingQueue<String> mailbox) throws Exception {
            for (int i = 0; i < 10; i++) {
                this.sum += i;
            }
            return new Integer(this.sum);
        }
 
    }
 
    @Test
    public void hoge2() throws InterruptedException, ExecutionException {
        JobButler jb = new JobButler(Executors.newCachedThreadPool());
        JobGateway<String, String> jgw1 = jb.submitNewJob("job1", new HogeJob("jon"));
        JobGateway<String, String> jgw2 = jb.submitNewJob("job2", new HogeJob("bob"));
        JobGateway<Integer, String> jgw3 = jb.submitNewJob("job3", new CountJob());
 
        assertEquals(jgw1.future.get(), "jon:0123456789");
        assertEquals(jgw2.future.get(), "bob:0123456789");
        assertEquals(jgw3.future.get(), new Integer(45));
 
        assertTrue(jgw1.future.isDone());
        assertTrue(jgw2.future.isDone());
        assertTrue(jgw3.future.isDone());
 
        assertNull(jb.getJobGateway("job1"));
        assertNull(jb.getJobGateway("job2"));
        assertNull(jb.getJobGateway("job3"));
 
        jb.shutdown();
    }
 
    class SlowCountJob implements JobHandle<Integer, String> {
 
        private int sum = 0;
        private final String name;
 
        public SlowCountJob(String name) {
            this.name = name;
        }
 
        @Override
        public Integer main(BlockingQueue<String> mailbox) throws Exception {
            int i = 0;
            while (!"terminate".equals(mailbox.poll(1000, TimeUnit.MILLISECONDS))) {
                this.sum += i;
                i++;
                System.out.println(this.name + ":" + this.sum);
            }
            return new Integer(this.sum);
        }
 
    }
 
    @Test(expectedExceptions = CancellationException.class)
    public void hoge3() throws InterruptedException, ExecutionException, TimeoutException {
        JobButler jb = new JobButler(Executors.newCachedThreadPool());
        JobGateway<Integer, String> jgw = jb.submitNewJob("job1", new SlowCountJob("scj-1"));
 
        try {
            Thread.sleep(5000);
        } catch (InterruptedException ignore) {
        }
        // cancel flag was set, but not interrupted ... orphan thread!!!
        jgw.future.cancel(false);
        assertTrue(jgw.future.isCancelled());
        assertTrue(jgw.future.isDone());
        try {
            jgw.future.get(1, TimeUnit.SECONDS);
        } finally {
            // cancel(false) internally called done().
            assertNull(jb.getJobGateway("job1"));
            jb.shutdown();
        }
    }
 
    @Test(expectedExceptions = CancellationException.class)
    public void hoge4() throws InterruptedException, ExecutionException, TimeoutException {
        JobButler jb = new JobButler(Executors.newCachedThreadPool());
        JobGateway<Integer, String> jgw = jb.submitNewJob("job1", new SlowCountJob("scj-2"));
 
        try {
            Thread.sleep(5000);
        } catch (InterruptedException ignore) {
        }
        // cancel flag was set, and, inerrupted -> thread terminates.
        jgw.future.cancel(true);
        assertTrue(jgw.future.isCancelled());
        assertTrue(jgw.future.isDone());
        try {
            jgw.future.get(1, TimeUnit.SECONDS);
        } finally {
            assertNull(jb.getJobGateway("job1"));
            jb.shutdown();
        }
    }
 
    class AbnormalEndCountJob implements JobHandle<Integer, String> {
 
        private int sum = 0;
        private final String name;
        private final int stopper;
 
        public AbnormalEndCountJob(String name, int stopper) {
            this.name = name;
            this.stopper = stopper;
        }
 
        @Override
        public Integer main(BlockingQueue<String> mailbox) throws Exception {
            int i = 0;
            while (!"terminate".equals(mailbox.poll(1000, TimeUnit.MILLISECONDS))) {
                this.sum += i;
                i++;
                System.out.println(this.name + ":" + this.sum);
                if (i > stopper) {
                    throw new IllegalStateException("stopper threashold over.");
                }
            }
            return new Integer(this.sum);
        }
 
    }
 
    @Test(expectedExceptions = IllegalStateException.class)
    public void hoge5() throws InterruptedException, ExecutionException, TimeoutException, Throwable {
        JobButler jb = new JobButler(Executors.newCachedThreadPool());
        JobGateway<Integer, String> jgw = jb.submitNewJob("job1", new AbnormalEndCountJob("abend-job-1", 3));
 
        try {
            Thread.sleep(5000);
        } catch (InterruptedException ignore) {
        }
        // job is already done (stopper = 3 means "After 3 seconds, throw IllegalStateException")
        // so, cancel() does not effect any more. isCancelled() returns false.
        jgw.future.cancel(true);
        assertFalse(jgw.future.isCancelled());
        assertTrue(jgw.future.isDone());
        try {
            jgw.future.get(1, TimeUnit.SECONDS);
        } catch (ExecutionException expected) {
            Throwable t = expected.getCause();
            assertEquals(t.getMessage(), "stopper threashold over.");
            throw t;
        } finally {
            assertNull(jb.getJobGateway("job1"));
            jb.shutdown();
        }
    }
 
    class TypicalMessageLoopJob implements JobHandle<String, String> {
 
        @Override
        public String main(BlockingQueue<String> mailbox) throws Exception {
            StringBuilder sb = new StringBuilder(1024);
            while (true) {
                String message = mailbox.poll();
                if ("__END__".equals(message)) {
                    break;
                }
                sb.append(message);
            }
            return sb.toString();
        }
    }
 
    @Test
    public void hoge6() throws InterruptedException, ExecutionException, TimeoutException, Throwable {
        JobButler jb = new JobButler(Executors.newCachedThreadPool());
        JobGateway<String, String> jgw = jb.submitNewJob("job1", new TypicalMessageLoopJob());
 
        jgw.mailbox.put("hello, ");
        jgw.mailbox.put("bob. ");
        jgw.mailbox.put("Good ");
        jgw.mailbox.put("Morning!");
        jgw.mailbox.put("__END__");
        jgw.mailbox.put("(this message must be ignored.");
        assertEquals(jgw.future.get(), "hello, bob. Good Morning!");
        assertFalse(jgw.future.isCancelled());
        assertTrue(jgw.future.isDone());
        try {
            Thread.sleep(1000);
        } catch (InterruptedException ignore) {}
        assertNull(jb.getJobGateway("job1"));
        jb.shutdown();
    }
 
}