下面列出了java.util.concurrent.atomic.AtomicInteger#wait() 实例代码，或者点击链接到github查看源代码，也可以在右侧发表评论。
public static List<ParseObject> getAllObjects (final String objectClass,int startLimit){
ParseQuery<ParseObject> pq = ParseQuery.getQuery(objectClass);
pq.limit(startLimit);
AtomicInteger mutex = new AtomicInteger(0);
List<ParseObject> allObjects = new ArrayList<ParseObject>();
pq.findInBackground(privateGetAllObjects(objectClass,pq.getLimit(),0,mutex,allObjects));
synchronized(mutex){
try {
if(mutex.compareAndSet(0,0))
mutex.wait();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
return allObjects;
}
private static void waitForCondition(final AtomicInteger eventCount, int EXPECTED_EVENT_COUNT,
long timeout)
{
synchronized (eventCount) {
if (eventCount.get() != EXPECTED_EVENT_COUNT) {
try {
eventCount.wait(timeout);
} catch (InterruptedException e) {
System.out.println("Interrupted unexpectedly!");
throw new RuntimeException(e);
}
}
}
}
private static void waitForCondition(final AtomicInteger eventCount, int EXPECTED_EVENT_COUNT,
long timeout)
{
synchronized (eventCount) {
if (eventCount.get() != EXPECTED_EVENT_COUNT) {
try {
eventCount.wait(timeout);
} catch (InterruptedException e) {
System.out.println("Interrupted unexpectedly!");
throw new RuntimeException(e);
}
}
}
}
/**
* Find all datasets inside an n5 container
* A dataset is any one of:
* - N5 dataset
* - multi-sclae group
* - paintera dataset
* @param n5 container
* @param keepLooking discover datasets while while {@code keepLooking.get() == true}
* @param es ExecutorService for parallelization of discovery
* @return List of all contained datasets (paths wrt to the root of the container)
*/
public static List<String> discoverDatasets(
final N5Reader n5,
final BooleanSupplier keepLooking,
final ExecutorService es)
{
final List<String> datasets = new ArrayList<>();
final AtomicInteger counter = new AtomicInteger(1);
Future<?> f = es.submit(() -> discoverSubdirectories(n5, "", datasets, es, counter, keepLooking));
while (true) {
try {
synchronized (counter) {
counter.wait();
final int count = counter.get();
LOG.debug("Current discovery task count is {}", count);
if (counter.get() <= 0) {
LOG.debug("Finished all discovery tasks.");
break;
}
}
} catch (InterruptedException e) {
LOG.debug("Was interrupted -- will stop dataset discovery.");
Thread.currentThread().interrupt();
break;
}
}
Collections.sort(datasets);
return datasets;
}
private static void waitForCondition(final AtomicInteger eventCount, int EXPECTED_EVENT_COUNT,
long timeout)
{
synchronized (eventCount) {
if (eventCount.get() != EXPECTED_EVENT_COUNT) {
try {
eventCount.wait(timeout);
} catch (InterruptedException e) {
System.out.println("Interrupted unexpectedly!");
throw new RuntimeException(e);
}
}
}
}
private static void waitForCondition(final AtomicInteger eventCount, int EXPECTED_EVENT_COUNT,
long timeout)
{
synchronized (eventCount) {
if (eventCount.get() != EXPECTED_EVENT_COUNT) {
try {
eventCount.wait(timeout);
} catch (InterruptedException e) {
System.out.println("Interrupted unexpectedly!");
throw new RuntimeException(e);
}
}
}
}
private static void waitForCondition(final AtomicInteger eventCount, int EXPECTED_EVENT_COUNT,
long timeout)
{
synchronized (eventCount) {
if (eventCount.get() != EXPECTED_EVENT_COUNT) {
try {
eventCount.wait(timeout);
} catch (InterruptedException e) {
System.out.println("Interrupted unexpectedly!");
throw new RuntimeException(e);
}
}
}
}
private static void waitForCondition(final AtomicInteger eventCount, int EXPECTED_EVENT_COUNT,
long timeout)
{
synchronized (eventCount) {
if (eventCount.get() != EXPECTED_EVENT_COUNT) {
try {
eventCount.wait(timeout);
} catch (InterruptedException e) {
System.out.println("Interrupted unexpectedly!");
throw new RuntimeException(e);
}
}
}
}
private static void waitForCondition(final AtomicInteger eventCount, int EXPECTED_EVENT_COUNT,
long timeout)
{
synchronized (eventCount) {
if (eventCount.get() != EXPECTED_EVENT_COUNT) {
try {
eventCount.wait(timeout);
} catch (InterruptedException e) {
System.out.println("Interrupted unexpectedly!");
throw new RuntimeException(e);
}
}
}
}
public static void login (String userClass,String userNameKey, String userName, String passwordKey, String password) throws LoginException{
AtomicInteger loged = new AtomicInteger(0);
Map<String,Object> kval = new HashMap<>();
kval.put(userNameKey, userName);
checkExsistance(userClass, kval, new GetCallback<ParseObject>() {
@Override
public void done(ParseObject arg0, ParseException arg1) {
synchronized(loged){
loged.compareAndSet(0, arg0 == null ? 3 : arg0.get(passwordKey).equals(password) ? 1 : 2);
loged.notifyAll();
}
}
});
synchronized(loged){
try {
if(loged.compareAndSet(0,0))
loged.wait();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
switch (loged.get()) {
case 2:{
LOGGER.severe("password doesn't match");
throw new LoginException("password doesn't match"); }
case 3:{
LOGGER.severe("user doesn't exists");
throw new LoginException("user doesn't exists");}
default:
break;
}
}
private static void waitForCondition(final AtomicInteger eventCount, int EXPECTED_EVENT_COUNT,
long timeout)
{
synchronized (eventCount) {
if (eventCount.get() != EXPECTED_EVENT_COUNT) {
try {
eventCount.wait(timeout);
} catch (InterruptedException e) {
System.out.println("Interrupted unexpectedly!");
throw new RuntimeException(e);
}
}
}
}
private static void waitForCondition(final AtomicInteger eventCount, int EXPECTED_EVENT_COUNT,
long timeout)
{
synchronized (eventCount) {
if (eventCount.get() != EXPECTED_EVENT_COUNT) {
try {
eventCount.wait(timeout);
} catch (InterruptedException e) {
System.out.println("Interrupted unexpectedly!");
throw new RuntimeException(e);
}
}
}
}
/**
* Main method used for sorting from clients. Input is sorted in place using multiple threads.
*
* @param input The array to sort.
* @param <T> The type of the objects being sorted, must extend Comparable.
*/
public static <T extends Comparable<T>> void quicksort(T[] input) {
final AtomicInteger count = new AtomicInteger(1);
pool.execute(new QuicksortRunnable<T>(input, 0, input.length - 1, count));
try {
synchronized (count) {
count.wait();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
protected boolean sendValues(Iterable<IN> values, long checkpointId, long timestamp) throws Exception {
final AtomicInteger updatesCount = new AtomicInteger(0);
final AtomicInteger updatesConfirmed = new AtomicInteger(0);
final AtomicReference<Throwable> exception = new AtomicReference<>();
FutureCallback<ResultSet> callback = new FutureCallback<ResultSet>() {
@Override
public void onSuccess(ResultSet resultSet) {
updatesConfirmed.incrementAndGet();
if (updatesCount.get() > 0) { // only set if all updates have been sent
if (updatesCount.get() == updatesConfirmed.get()) {
synchronized (updatesConfirmed) {
updatesConfirmed.notifyAll();
}
}
}
}
@Override
public void onFailure(Throwable throwable) {
if (exception.compareAndSet(null, throwable)) {
LOG.error("Error while sending value.", throwable);
synchronized (updatesConfirmed) {
updatesConfirmed.notifyAll();
}
}
}
};
//set values for prepared statement
int updatesSent = 0;
for (IN value : values) {
for (int x = 0; x < value.getArity(); x++) {
fields[x] = value.getField(x);
}
//insert values and send to cassandra
BoundStatement s = preparedStatement.bind(fields);
s.setDefaultTimestamp(timestamp);
ResultSetFuture result = session.executeAsync(s);
updatesSent++;
if (result != null) {
//add callback to detect errors
Futures.addCallback(result, callback);
}
}
updatesCount.set(updatesSent);
synchronized (updatesConfirmed) {
while (exception.get() == null && updatesSent != updatesConfirmed.get()) {
updatesConfirmed.wait();
}
}
if (exception.get() != null) {
LOG.warn("Sending a value failed.", exception.get());
return false;
} else {
return true;
}
}
@Override
protected boolean sendValues(Iterable<Row> values, long checkpointId, long timestamp) throws Exception {
final AtomicInteger updatesCount = new AtomicInteger(0);
final AtomicInteger updatesConfirmed = new AtomicInteger(0);
final AtomicReference<Throwable> exception = new AtomicReference<>();
FutureCallback<ResultSet> callback = new FutureCallback<ResultSet>() {
@Override
public void onSuccess(ResultSet resultSet) {
updatesConfirmed.incrementAndGet();
if (updatesCount.get() > 0) { // only set if all updates have been sent
if (updatesCount.get() == updatesConfirmed.get()) {
synchronized (updatesConfirmed) {
updatesConfirmed.notifyAll();
}
}
}
}
@Override
public void onFailure(Throwable throwable) {
if (exception.compareAndSet(null, throwable)) {
LOG.error("Error while sending value.", throwable);
synchronized (updatesConfirmed) {
updatesConfirmed.notifyAll();
}
}
}
};
//set values for prepared statement
int updatesSent = 0;
for (Row value : values) {
for (int x = 0; x < arity; x++) {
fields[x] = value.getField(x);
}
//insert values and send to cassandra
BoundStatement s = preparedStatement.bind(fields);
s.setDefaultTimestamp(timestamp);
ResultSetFuture result = session.executeAsync(s);
updatesSent++;
if (result != null) {
//add callback to detect errors
Futures.addCallback(result, callback);
}
}
updatesCount.set(updatesSent);
synchronized (updatesConfirmed) {
while (exception.get() == null && updatesSent != updatesConfirmed.get()) {
updatesConfirmed.wait();
}
}
if (exception.get() != null) {
LOG.warn("Sending a value failed.", exception.get());
return false;
} else {
return true;
}
}
@Override
protected boolean sendValues(Iterable<IN> values, long checkpointId, long timestamp) throws Exception {
final AtomicInteger updatesCount = new AtomicInteger(0);
final AtomicInteger updatesConfirmed = new AtomicInteger(0);
final AtomicReference<Throwable> exception = new AtomicReference<>();
FutureCallback<ResultSet> callback = new FutureCallback<ResultSet>() {
@Override
public void onSuccess(ResultSet resultSet) {
updatesConfirmed.incrementAndGet();
if (updatesCount.get() > 0) { // only set if all updates have been sent
if (updatesCount.get() == updatesConfirmed.get()) {
synchronized (updatesConfirmed) {
updatesConfirmed.notifyAll();
}
}
}
}
@Override
public void onFailure(Throwable throwable) {
if (exception.compareAndSet(null, throwable)) {
LOG.error("Error while sending value.", throwable);
synchronized (updatesConfirmed) {
updatesConfirmed.notifyAll();
}
}
}
};
//set values for prepared statement
int updatesSent = 0;
for (IN value : values) {
for (int x = 0; x < value.getArity(); x++) {
fields[x] = value.getField(x);
}
//insert values and send to cassandra
BoundStatement s = preparedStatement.bind(fields);
s.setDefaultTimestamp(timestamp);
ResultSetFuture result = session.executeAsync(s);
updatesSent++;
if (result != null) {
//add callback to detect errors
Futures.addCallback(result, callback);
}
}
updatesCount.set(updatesSent);
synchronized (updatesConfirmed) {
while (exception.get() == null && updatesSent != updatesConfirmed.get()) {
updatesConfirmed.wait();
}
}
if (exception.get() != null) {
LOG.warn("Sending a value failed.", exception.get());
return false;
} else {
return true;
}
}
@Override
protected boolean sendValues(Iterable<Row> values, long checkpointId, long timestamp) throws Exception {
final AtomicInteger updatesCount = new AtomicInteger(0);
final AtomicInteger updatesConfirmed = new AtomicInteger(0);
final AtomicReference<Throwable> exception = new AtomicReference<>();
FutureCallback<ResultSet> callback = new FutureCallback<ResultSet>() {
@Override
public void onSuccess(ResultSet resultSet) {
updatesConfirmed.incrementAndGet();
if (updatesCount.get() > 0) { // only set if all updates have been sent
if (updatesCount.get() == updatesConfirmed.get()) {
synchronized (updatesConfirmed) {
updatesConfirmed.notifyAll();
}
}
}
}
@Override
public void onFailure(Throwable throwable) {
if (exception.compareAndSet(null, throwable)) {
LOG.error("Error while sending value.", throwable);
synchronized (updatesConfirmed) {
updatesConfirmed.notifyAll();
}
}
}
};
//set values for prepared statement
int updatesSent = 0;
for (Row value : values) {
for (int x = 0; x < arity; x++) {
fields[x] = value.getField(x);
}
//insert values and send to cassandra
BoundStatement s = preparedStatement.bind(fields);
s.setDefaultTimestamp(timestamp);
ResultSetFuture result = session.executeAsync(s);
updatesSent++;
if (result != null) {
//add callback to detect errors
Futures.addCallback(result, callback);
}
}
updatesCount.set(updatesSent);
synchronized (updatesConfirmed) {
while (exception.get() == null && updatesSent != updatesConfirmed.get()) {
updatesConfirmed.wait();
}
}
if (exception.get() != null) {
LOG.warn("Sending a value failed.", exception.get());
return false;
} else {
return true;
}
}
protected void startProducers(final ConnectionFactory factory,
final Destination dest,
final int msgCount) throws Exception {
// Use concurrent send
if (useConcurrentSend) {
final AtomicInteger producerLock = new AtomicInteger(producerCount);
for (int i = 0; i < producerCount; i++) {
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
sendMessages(factory.createConnection(), dest, msgCount);
} catch (Exception e) {
e.printStackTrace();
}
synchronized (producerLock) {
producerLock.decrementAndGet();
producerLock.notifyAll();
}
}
});
t.start();
}
// Wait for all producers to finish sending
synchronized (producerLock) {
while (producerLock.get() != 0) {
producerLock.wait(2000);
}
}
// Use serialized send
} else {
for (int i = 0; i < producerCount; i++) {
sendMessages(factory.createConnection(), dest, msgCount);
}
}
}
@Override
protected boolean sendValues(Iterable<IN> values, long checkpointId, long timestamp) throws Exception {
final AtomicInteger updatesCount = new AtomicInteger(0);
final AtomicInteger updatesConfirmed = new AtomicInteger(0);
final AtomicReference<Throwable> exception = new AtomicReference<>();
FutureCallback<ResultSet> callback = new FutureCallback<ResultSet>() {
@Override
public void onSuccess(ResultSet resultSet) {
updatesConfirmed.incrementAndGet();
if (updatesCount.get() > 0) { // only set if all updates have been sent
if (updatesCount.get() == updatesConfirmed.get()) {
synchronized (updatesConfirmed) {
updatesConfirmed.notifyAll();
}
}
}
}
@Override
public void onFailure(Throwable throwable) {
if (exception.compareAndSet(null, throwable)) {
LOG.error("Error while sending value.", throwable);
synchronized (updatesConfirmed) {
updatesConfirmed.notifyAll();
}
}
}
};
//set values for prepared statement
int updatesSent = 0;
for (IN value : values) {
for (int x = 0; x < value.getArity(); x++) {
fields[x] = value.getField(x);
}
//insert values and send to cassandra
BoundStatement s = preparedStatement.bind(fields);
s.setDefaultTimestamp(timestamp);
ResultSetFuture result = session.executeAsync(s);
updatesSent++;
if (result != null) {
//add callback to detect errors
Futures.addCallback(result, callback);
}
}
updatesCount.set(updatesSent);
synchronized (updatesConfirmed) {
while (exception.get() == null && updatesSent != updatesConfirmed.get()) {
updatesConfirmed.wait();
}
}
if (exception.get() != null) {
LOG.warn("Sending a value failed.", exception.get());
return false;
} else {
return true;
}
}
/**
* Returns a link for the remote address if cached; otherwise opens, caches and returns.
* When it opens a link for the remote address, only one attempt for the address is made at a given time
*
* @param remoteAddr the remote socket address
* @param encoder the encoder
* @param listener the link listener
* @return a link associated with the address
*/
@Override
public <T> Link<T> open(final SocketAddress remoteAddr, final Encoder<? super T> encoder,
final LinkListener<? super T> listener) throws IOException {
Link<T> link = null;
for (int i = 0; i <= this.numberOfTries; ++i) {
LinkReference linkRef = this.addrToLinkRefMap.get(remoteAddr);
if (linkRef != null) {
link = (Link<T>) linkRef.getLink();
if (LOG.isLoggable(Level.FINE)) {
LOG.log(Level.FINE, "Link {0} for {1} found", new Object[]{link, remoteAddr});
}
if (link != null) {
return link;
}
}
if (i == this.numberOfTries) {
// Connection failure
throw new ConnectException("Connection to " + remoteAddr + " refused");
}
LOG.log(Level.FINE, "No cached link for {0} thread {1}",
new Object[]{remoteAddr, Thread.currentThread()});
// no linkRef
final LinkReference newLinkRef = new LinkReference();
final LinkReference prior = this.addrToLinkRefMap.putIfAbsent(remoteAddr, newLinkRef);
final AtomicInteger flag = prior != null ?
prior.getConnectInProgress() : newLinkRef.getConnectInProgress();
synchronized (flag) {
if (!flag.compareAndSet(0, 1)) {
while (flag.get() == 1) {
try {
flag.wait();
} catch (final InterruptedException ex) {
LOG.log(Level.WARNING, "Wait interrupted", ex);
}
}
}
}
linkRef = this.addrToLinkRefMap.get(remoteAddr);
link = (Link<T>) linkRef.getLink();
if (link != null) {
return link;
}
ChannelFuture connectFuture = null;
try {
connectFuture = this.clientBootstrap.connect(remoteAddr);
connectFuture.syncUninterruptibly();
link = new NettyLink<>(connectFuture.channel(), encoder, listener);
linkRef.setLink(link);
synchronized (flag) {
flag.compareAndSet(1, 2);
flag.notifyAll();
}
break;
} catch (final Exception e) {
if (e instanceof ConnectException) {
LOG.log(Level.WARNING, "Connection refused. Retry {0} of {1}",
new Object[]{i + 1, this.numberOfTries});
synchronized (flag) {
flag.compareAndSet(1, 0);
flag.notifyAll();
}
if (i < this.numberOfTries) {
try {
Thread.sleep(retryTimeout);
} catch (final InterruptedException interrupt) {
LOG.log(Level.WARNING, "Thread {0} interrupted while sleeping", Thread.currentThread());
}
}
} else {
throw e;
}
}
}
return link;
}